fix: extend shared memory

Author: darthunix

executor/src/server.rs

@@ -583,6 +583,9 @@ fn start_data_flow(conn: &mut Connection, storage: &mut Storage) -> Result<TaskR
     let ctx = state.task_ctx();
 
     let conn_id = conn.id;
+    let client_pid_val = conn
+        .client_pid
+        .load(std::sync::atomic::Ordering::Relaxed); // snapshot pid
     let handle = tokio::spawn(async move {
         // Execute a single partition; PgScanExec uses a single stream
         match plan.execute(0, ctx) {
@@ -602,7 +605,13 @@ fn start_data_flow(conn: &mut Connection, storage: &mut Storage) -> Result<TaskR
                             if pg_attrs.is_empty() {
                                 continue;
                             }
-                            let _ = encode_and_write_rows(&batch, &pg_attrs, &mut ring);
+                            let wrote = encode_and_write_rows(&batch, &pg_attrs, &mut ring);
+                            if wrote > 0 {
+                                let pid = client_pid_val;
+                                if pid > 0 && pid != i32::MAX {
+                                    unsafe { libc::kill(pid as libc::pid_t, libc::SIGUSR1) };
+                                }
+                            }
                         }
                     }
                 }

postgres/src/lib.rs

@@ -89,10 +89,27 @@ unsafe extern "C-unwind" fn pg_fusion_shmem_request_hook() {
         .layout
         .size();
 
+    // Slot blocks (per-connection heap page buffers)
+    let blksz = pgrx::pg_sys::BLCKSZ as usize;
+    let slot_blocks_layout = executor::layout::slot_blocks_layout(
+        worker::SLOTS_PER_CONN,
+        blksz,
+        worker::BLOCKS_PER_SLOT,
+    )
+    .expect("slot_blocks_layout");
+    let slot_blocks_sz = slot_blocks_layout.layout.size() * num;
+
+    // Per-connection result ring buffers
+    let result_ring_layout =
+        executor::layout::result_ring_layout(worker::RESULT_RING_CAP).expect("result_ring_layout");
+    let result_ring_sz = result_ring_layout.layout.size() * num;
+
     let total = flags_sz
         .saturating_add(conns_sz)
         .saturating_add(stack_sz)
-        .saturating_add(pid_sz);
+        .saturating_add(pid_sz)
+        .saturating_add(slot_blocks_sz)
+        .saturating_add(result_ring_sz);
 
     pgrx::pg_sys::RequestAddinShmemSpace(total);
 }

postgres/src/worker.rs

@@ -18,9 +18,9 @@ const TOKIO_THREAD_NUMBER: usize = 1;
 pub(crate) const RECV_CAP: usize = 8192;
 pub(crate) const SEND_CAP: usize = 8192;
 // Heap block buffer configuration per connection
-const SLOTS_PER_CONN: usize = 2; // two logical slots per connection
-const BLOCKS_PER_SLOT: usize = 2; // double buffering inside each slot
-const RESULT_RING_CAP: usize = 64 * 1024; // bytes per-connection for result rows
+pub(crate) const SLOTS_PER_CONN: usize = 2; // two logical slots per connection
+pub(crate) const BLOCKS_PER_SLOT: usize = 2; // double buffering inside each slot
+pub(crate) const RESULT_RING_CAP: usize = 64 * 1024; // bytes per-connection for result rows
 
 #[pg_guard]
 pub(crate) unsafe extern "C-unwind" fn init_datafusion_worker() {