refragment simulation snapshots to ensure client frames advance atomically

src/pp/pp.h

@@ -1,3 +1,5 @@
+#define P_MaxFrameSnapshotFragments Kibi(4)
+
 ////////////////////////////////////////////////////////////
 //~ Key types
 
@@ -240,6 +242,13 @@ Struct(P_Frame)
   i64 max_constraints;
   i64 constraints_count;
   P_Constraint *constraints;
+
+  //////////////////////////////
+  //- Snapshot-assembly state
+
+  u64 fragments_count;
+  u64 received_fragments_count;
+  u64 received_fragment_bits[(P_MaxFrameSnapshotFragments + 63) / 64];
 };
 
 Struct(P_FrameBin)

src/pp/pp_sim/pp_sim_core.c

@@ -861,7 +861,7 @@ void S_TickForever(WaveLaneCtx *lane)
         PackedDeltaNode *next;
         String packed;
       };
-      i64 total_delta_bytes = 0;
+      i64 total_delta_accum = 0;
       PackedDeltaNode *first_delta_node = 0;
       PackedDeltaNode *last_delta_node = 0;
 
@@ -887,30 +887,49 @@ void S_TickForever(WaveLaneCtx *lane)
             PackedDeltaNode *delta_node = PushStruct(frame_arena, PackedDeltaNode);
             delta_node->packed = PushString(frame_arena, packed);
             SllQueuePush(first_delta_node, last_delta_node, delta_node);
-            total_delta_bytes += packed.len;
+            total_delta_accum += packed.len;
           }
         }
       }
 
-      //- Collect & send snapshots
+      //- Count fragments
+      i64 fragment_delta_accum_cutoff_threshold = NET_PacketSize / 2;
+      i64 fragments_count = 0;
+      {
+        i64 cur_fragment_delta_accum = 0;
+        for (PackedDeltaNode *delta_node = first_delta_node; delta_node; delta_node = delta_node->next)
+        {
+          cur_fragment_delta_accum += delta_node->packed.len;
+          if (!delta_node->next || (i64)(cur_fragment_delta_accum + delta_node->next->packed.len) >= fragment_delta_accum_cutoff_threshold)
+          {
+            fragments_count += 1;
+            cur_fragment_delta_accum = 0;
+          }
+        }
+      }
+
+      //- Collect & send fragments
       {
-        u64 snapshot_cutoff_threshold = NET_PacketSize / 2;
         PackedDeltaNode *delta_node = first_delta_node;
 
-        u64 snapshot_start = 0;
+        u64 fragment_start = 0;
-        b32 new_snapshot = 1;
+        i64 fragment_idx = 0;
+        i64 cur_fragment_delta_accum = 0;
+        b32 new_fragment = 1;
         b32 done = 0;
         while (!done)
         {
           PackedDeltaNode *next_delta_node = delta_node ? delta_node->next : 0;
           {
-            //- Init snapshot
+            //- Init fragment
-            if (new_snapshot)
+            if (new_fragment)
             {
-              new_snapshot = 0;
+              new_fragment = 0;
               BB_ResetWriter(&packer_bbw);
-              snapshot_start = BB_GetNumBytesWritten(&packer_bbw);
+              fragment_start = BB_GetNumBytesWritten(&packer_bbw);
               BB_WriteBit(&packer_bbw, 1); // Raw
+              BB_WriteIV(&packer_bbw, fragment_idx);
+              BB_WriteIV(&packer_bbw, fragments_count);
               BB_WriteIV(&packer_bbw, src_frame->tick);
               BB_WriteIV(&packer_bbw, world_frame->tick);
               BB_WriteIV(&packer_bbw, world_frame->time_ns);
@@ -924,43 +943,45 @@ void S_TickForever(WaveLaneCtx *lane)
             if (delta_node)
             {
               BB_WriteBytes(&packer_bbw, delta_node->packed);
+              cur_fragment_delta_accum += delta_node->packed.len;
             }
 
-            //- Submit snapshot
+            //- Submit fragment
             {
               if (!delta_node || !next_delta_node)
               {
-                new_snapshot = 1;
+                new_fragment = 1;
                 done = 1;
               }
-              u64 next_len = 0;
+              i64 next_delta_size = 0;
               if (next_delta_node)
               {
-                next_len = next_delta_node->packed.len;
+                next_delta_size = next_delta_node->packed.len;
               }
-              u64 cur_snapshot_len = BB_GetNumBytesWritten(&packer_bbw);
+              if (cur_fragment_delta_accum + next_delta_size >= fragment_delta_accum_cutoff_threshold)
-              if ((cur_snapshot_len - snapshot_start) + next_len >= snapshot_cutoff_threshold)
               {
-                new_snapshot = 1;
+                new_fragment = 1;
               }
             }
-            if (new_snapshot)
+            if (new_fragment)
             {
-              u64 snapshot_end = BB_GetNumBytesWritten(&packer_bbw);
+              u64 fragment_end = BB_GetNumBytesWritten(&packer_bbw);
-              String snapshot = Zi;
+              String fragment = Zi;
-              snapshot.text = BB_GetWrittenRaw(&packer_bbw) + snapshot_start;
+              fragment.text = BB_GetWrittenRaw(&packer_bbw) + fragment_start;
-              snapshot.len = snapshot_end - snapshot_start;
+              fragment.len = fragment_end - fragment_start;
-              NET_Send(net_pipe, client->net_key, snapshot, NET_SendFlag_Raw);
+              NET_Send(net_pipe, client->net_key, fragment, NET_SendFlag_Raw);
 
               // Sanity check to catch whenever we end up packing too much information
-              // into a single raw snapshot, causing it to drop. This isn't necessarily
+              // into a single raw fragment, causing it to drop. This isn't necessarily
               // an error, but signals we may have botched the packing code or need tighter
               // compression
-              Assert(snapshot.len <= NET_PacketSize);
+              Assert(fragment.len <= NET_PacketSize);
+              fragment_idx += 1;
             }
           }
           delta_node = next_delta_node;
         }
+        Assert(fragment_idx == fragments_count);
       }
     }
 

src/pp/pp_vis/pp_vis_core.c

@@ -371,7 +371,7 @@ void V_TickForever(WaveLaneCtx *lane)
   i64 remote_ack = 0;
   i64 prev_snapshot_sent_at_ns = 0;
 
-  i64 prev_known_sim_tick = 0;
+  i64 ready_sim_tick = 0;
   i64 known_sim_tick = 0;
 
   Vec2I32 tiles_dims = VEC2I32(P_TilesPitch, P_TilesPitch);
@@ -2872,6 +2872,8 @@ void V_TickForever(WaveLaneCtx *lane)
 
           //- Read header
           BB_ReadBit(&bbr); // Raw
+          i64 fragment_idx = BB_ReadIV(&bbr);
+          i64 fragments_count = BB_ReadIV(&bbr);
           i64 src_tick = BB_ReadIV(&bbr);
           i64 dst_tick = BB_ReadIV(&bbr);
           i64 time_ns = BB_ReadIV(&bbr);
@@ -2886,6 +2888,8 @@ void V_TickForever(WaveLaneCtx *lane)
             remote_ack = tmp_remote_ack;
           }
 
+          if (dst_tick > ready_sim_tick && fragments_count < P_MaxFrameSnapshotFragments)
+          {
             P_Frame *src_frame = P_FrameFromTick(sim_world, src_tick);
             if (src_frame->tick == src_tick)
             {
@@ -2897,9 +2901,20 @@ void V_TickForever(WaveLaneCtx *lane)
               {
                 dst_frame = P_PushFrame(sim_world, src_frame, dst_tick);
               }
+              dst_frame->fragments_count = fragments_count;
               dst_frame->time_ns = time_ns;
               sim_world->seed = world_seed;
 
+              if (fragment_idx < P_MaxFrameSnapshotFragments)
+              {
+                u64 *frag_bit_chunk = &dst_frame->received_fragment_bits[fragment_idx / 64];
+                u64 frag_bit = (u64)1 << fragment_idx;
+                b32 is_new = !(*frag_bit_chunk & frag_bit);
+                if (is_new)
+                {
+                  *frag_bit_chunk |= frag_bit;
+                  dst_frame->received_fragments_count += 1;
+
                   //- Read deltas
                   BB_ReadAlignToNextByte(&bbr);
 
@@ -2923,10 +2938,20 @@ void V_TickForever(WaveLaneCtx *lane)
                       done = 1;
                     }
                   }
+
+                  // Mark snapshot as ready if assembled
+                  if (dst_frame->received_fragments_count >= dst_frame->fragments_count)
+                  {
+                    ready_sim_tick = dst_tick;
                   }
                 }
               }
             }
+          }
+
+        }
+      }
+    }
 
     //////////////////////////////
     //- Apply sim msgs
@@ -2973,7 +2998,7 @@ void V_TickForever(WaveLaneCtx *lane)
 
     // TODO: Remove this
     // TODO: Delete all frames except for prediction base & remote ack
-    P_ClearFrames(sim_world, I64Min, known_sim_tick - SIM_TICKS_PER_SECOND);
+    P_ClearFrames(sim_world, I64Min, ready_sim_tick - SIM_TICKS_PER_SECOND);
     // P_ClearFrames(sim_world, I64Min, sim_world->last_frame->tick - 1);
     P_Frame *sim_frame = sim_world->last_frame;
 
@@ -3031,9 +3056,7 @@ void V_TickForever(WaveLaneCtx *lane)
 
     // frame->predict_to = sim_world->last_frame->tick + MaxF64(CeilF64(ping * SIM_TICKS_PER_SECOND), 1.0);
 
-    // frame->predict_to = known_sim_tick + 10;
+    frame->predict_to = ready_sim_tick + 10;
-
-    frame->predict_to = known_sim_tick + 20;
 
     //////////////////////////////
     //- Create player control
@@ -3087,10 +3110,10 @@ void V_TickForever(WaveLaneCtx *lane)
         i64 total_delta_bytes = 0;
         PackedDeltaNode *first_delta_node = 0;
         PackedDeltaNode *last_delta_node = 0;
-        if (frame->predict_to >= known_sim_tick)
+        if (frame->predict_to >= ready_sim_tick)
         {
           i64 last_send_tick = frame->predict_to;
-          i64 first_send_tick = MaxI64(known_sim_tick, remote_ack + 1);
+          i64 first_send_tick = MaxI64(ready_sim_tick, remote_ack + 1);
           first_send_tick = MaxI64(first_send_tick, last_send_tick - max_control_sends);
           for (i64 send_tick = first_send_tick; send_tick <= last_send_tick; ++send_tick)
           {
@@ -3138,7 +3161,7 @@ void V_TickForever(WaveLaneCtx *lane)
                 BB_ResetWriter(&packer_bbw);
                 snapshot_start = BB_GetNumBytesWritten(&packer_bbw);
                 BB_WriteBit(&packer_bbw, 1); // Raw
-                BB_WriteIV(&packer_bbw, known_sim_tick);
+                BB_WriteIV(&packer_bbw, ready_sim_tick);
               }
 
               //- Append packed delta
@@ -3233,7 +3256,7 @@ void V_TickForever(WaveLaneCtx *lane)
       // FIXME: Not like this
       i64 max_predict_ticks = SIM_TICKS_PER_SECOND;
       last_predict_tick = frame->predict_to;
-      first_predict_tick = known_sim_tick - 2;
+      first_predict_tick = ready_sim_tick - 2;
       first_predict_tick = MaxI64(first_predict_tick, last_predict_tick - max_predict_ticks);
     }