Commit 11ccdc2d authored by Andrey Filippov's avatar Andrey Filippov

CLAUDE: 3-B rung B4 - pinned double-buffered bayer staging + async per-sensor H2D

Legacy tp_proc_set_image is a synchronous cudaMemcpy2D from pageable
JNA-marshalled memory (B0: 11.9 ms/scene, ~1.8 GB/s). New surface:
tp_proc_bayer_staging(slot 0/1) lazily allocates pinned staging
(num_cams*w*h floats) + a dedicated non-blocking copy stream + fence
event; tp_proc_set_image_staged enqueues the pinned->device copy of one
sensor and returns (sensor k DMAs while the host prepares k+1). All
resident-image consumers/writers fence (bayer_fence; kernel exec paths
already cuCtxSynchronize). Double buffer + non-blocking stream = the
RT-service inter-scene overlap provision. tp_proc_destroy drains the
copy stream before freeing pinned.

test_bayer_staged_jna (synthetic, cases.list 'bayer_staged'): both
slots x 16 sensors bit-exact readback + staged->legacy WAW ordering -
PASS. Full run_cases.sh ALL PASS (incl. pose_corr @tol 0).
Co-Authored-By: 's avatarClaude Fable 5 <noreply@anthropic.com>
parent ef36688d
...@@ -64,6 +64,7 @@ case peak test_peak ${REPO}/testdata/synth4/peak ...@@ -64,6 +64,7 @@ case peak test_peak ${REPO}/testdata/synth4/peak
case peak_recenter test_peak ${REPO}/testdata/synth4/peak_recenter --tol 1e-9 case peak_recenter test_peak ${REPO}/testdata/synth4/peak_recenter --tol 1e-9
case avg_td test_avg_td ${REPO}/testdata/synth4/avg_td --tol 0 case avg_td test_avg_td ${REPO}/testdata/synth4/avg_td --tol 0
case avg_td_oob test_avg_td_oob ${REPO}/testdata/synth4/avg_td_oob --tol 0 case avg_td_oob test_avg_td_oob ${REPO}/testdata/synth4/avg_td_oob --tol 0
case bayer_staged test_bayer_staged_jna ${REPO} --tol 0 # rung B4: synthetic, --data unused
#case convert_direct test_convert_direct ${REPO}/testdata/convert_direct_legacy # local blessed goldens (~200MB, make_convert_direct_case.py) #case convert_direct test_convert_direct ${REPO}/testdata/convert_direct_legacy # local blessed goldens (~200MB, make_convert_direct_case.py)
# gen <name(s)> <generator command, run from ${REPO}> # gen <name(s)> <generator command, run from ${REPO}>
......
// test_bayer_staged_jna.cu - validate the 3-B rung B4 staged bayer upload
// (libtileproc.so): pinned double-buffered staging + per-sensor async H2D on
// the dedicated non-blocking copy stream. Fully synthetic (no case data):
// 1. write a distinct pattern per sensor into staging slot 0, enqueue all 16
// async copies, read back via tp_proc_get_image (fences internally) ->
// must be BIT-EXACT (the copy path may never alter bits);
// 2. flip to slot 1 with a different pattern (double-buffer reuse) -> bit-exact;
// 3. WAW ordering: staged copy immediately chased by a legacy synchronous
// tp_proc_set_image of different data on the same sensor -> readback must
// equal the LEGACY data (the legacy path fences on in-flight staged copies).
// All comparisons tol 0 by design.
//
// Build (after jna/build_lib.sh):
// cd tile_processor_gpu && /usr/local/cuda-12.8/bin/nvcc -std=c++17 -O2 \
// -I src -I $HOME/git/cuda-samples/Common \
// jna/test_bayer_staged_jna.cu \
// -o tests_bin/test_bayer_staged_jna -L jna -ltileproc \
// -Xlinker -rpath -Xlinker $PWD/jna
// Run: tests_bin/test_bayer_staged_jna (cwd = repo root; no --data needed)
//
// Created on: Jul 16, 2026
// Author: Claude (Anthropic), for Elphel
#include <cstdio>
#include <cstring>
#include <cstdlib>
extern "C" {
void* tp_create_module(const char*, const char*);
const char* tp_last_error();
void* tp_proc_create(void*);
int tp_proc_setup(void*, int,int,int,int,int,int);
float* tp_proc_bayer_staging(void*, int);
int tp_proc_set_image_staged(void*, int, int);
int tp_proc_set_image(void*, int, const float*);
int tp_proc_get_image(void*, int, float*);
void tp_proc_destroy(void*);
void tp_destroy_module(void*);
}
static long compare_bits(const float* a, const float* b, size_t n){
long bad=0; for(size_t i=0;i<n;i++) if(memcmp(&a[i],&b[i],sizeof(float))) bad++; return bad;
}
static void fill_pattern(float* dst, int cam, int salt, size_t npix){
for(size_t i=0;i<npix;i++) dst[i]=(float)(cam*1000.0+salt*0.001+i*0.000123);
}
int main(int argc, char** argv){
const char* srcd = "src";
const char* devrt = "/usr/local/cuda-12.8/lib64/libcudadevrt.a";
for (int i = 1; i < argc; i++){
if (!strcmp(argv[i],"--src") && (i+1<argc)) srcd = argv[++i];
else if (!strcmp(argv[i],"--devrt") && (i+1<argc)) devrt = argv[++i];
else if (!strcmp(argv[i],"--data") && (i+1<argc)) ++i; // accepted for run_cases.sh uniformity (unused: synthetic)
else if (!strcmp(argv[i],"--tol") && (i+1<argc)) ++i; // tol-0 by design
else { fprintf(stderr,"Usage: %s [--src <kernel src>] [--devrt <libcudadevrt.a>]\n", argv[0]); return 1; }
}
const int NC=16, W=640, H=512; const size_t NPIX=(size_t)W*H;
void* m = tp_create_module(srcd, devrt);
if(!m){ printf("FAIL module: %s\n", tp_last_error()); return 1; }
void* p = tp_proc_create(m);
if(tp_proc_setup(p, NC, 1, W, H, 0, 0)){ printf("FAIL setup: %s\n", tp_last_error()); return 1; }
float* got = new float[NPIX];
long bad_total = 0;
// 1+2: both staging slots, all sensors, bit-exact readback
for(int slot=0; slot<2; slot++){
float* staging = tp_proc_bayer_staging(p, slot);
if(!staging){ printf("FAIL staging slot %d: %s\n", slot, tp_last_error()); return 1; }
for(int c=0;c<NC;c++){
fill_pattern(staging + (size_t)c*NPIX, c, slot+1, NPIX);
if(tp_proc_set_image_staged(p, c, slot)){ printf("FAIL set_image_staged cam %d slot %d: %s\n", c, slot, tp_last_error()); return 1; }
}
long bad_slot=0;
for(int c=0;c<NC;c++){
if(tp_proc_get_image(p, c, got)){ printf("FAIL get_image cam %d\n", c); return 1; }
bad_slot += compare_bits(got, staging + (size_t)c*NPIX, NPIX);
}
printf("slot %d: %d sensors x %zu px, bit-mismatches=%ld -> %s\n",
slot, NC, NPIX, bad_slot, bad_slot?"FAIL":"PASS");
bad_total += bad_slot;
}
// 3: WAW ordering - staged copy chased by legacy sync upload on the same cam
float* staging = tp_proc_bayer_staging(p, 0);
float* legacy = new float[NPIX];
fill_pattern(staging, 0, 7, NPIX); // staged data, cam 0
fill_pattern(legacy, 0, 9, NPIX); // different legacy data, cam 0
if(tp_proc_set_image_staged(p, 0, 0)){ printf("FAIL WAW staged: %s\n", tp_last_error()); return 1; }
if(tp_proc_set_image(p, 0, legacy)){ printf("FAIL WAW legacy\n"); return 1; }
if(tp_proc_get_image(p, 0, got)){ printf("FAIL WAW readback\n"); return 1; }
long bad_waw = compare_bits(got, legacy, NPIX);
printf("WAW staged->legacy: bit-mismatches vs legacy=%ld -> %s\n", bad_waw, bad_waw?"FAIL":"PASS");
bad_total += bad_waw;
tp_proc_destroy(p); tp_destroy_module(m);
delete [] got; delete [] legacy;
printf("test_bayer_staged_jna: %s\n", bad_total?"FAIL":"PASS (bit-exact)");
return bad_total?1:0;
}
...@@ -710,6 +710,19 @@ struct TpProc { ...@@ -710,6 +710,19 @@ struct TpProc {
float *gpu_pose_task_centers; float *gpu_pose_task_centers;
int pose_task_capacity; // template entries the buffers are sized for int pose_task_capacity; // template entries the buffers are sized for
int pose_task_count; // entries of the resident template (0 = none) int pose_task_count; // entries of the resident template (0 = none)
// 3-B rung B4 (design 2026-07-16): pinned double-buffered bayer staging +
// dedicated non-blocking copy stream. Java writes a whole sensor image into
// the pinned slot (one JNI region copy), tp_proc_set_image_staged enqueues
// the pinned->device cudaMemcpy2DAsync, so sensor k DMAs while the host
// prepares sensor k+1 (per-sensor copy/compute pipelining). Every consumer
// of the resident images fences on bayer_copy_event before touching them
// (kernel exec paths additionally cuCtxSynchronize before launch). The
// double buffer + non-blocking stream are the inter-scene overlap provision
// (RT service: upload scene N+1 while scene N solves). By Claude on 07/16/2026.
float *bayer_staging[2]; // pinned host, num_cams*img_w*img_h floats each (lazy)
cudaStream_t bayer_copy_stream;
cudaEvent_t bayer_copy_event;
bool bayer_copy_pending;
}; };
extern "C" { extern "C" {
...@@ -766,6 +779,8 @@ TpProc* tp_proc_create(TpModule* m){ ...@@ -766,6 +779,8 @@ TpProc* tp_proc_create(TpModule* m){
p->gpu_pose_dp_chain=nullptr; // By Claude on 07/16/2026 (DP spike) p->gpu_pose_dp_chain=nullptr; // By Claude on 07/16/2026 (DP spike)
p->gpu_pose_task_map=nullptr; p->gpu_pose_task_centers=nullptr; // By Claude on 07/16/2026 (rung B1) p->gpu_pose_task_map=nullptr; p->gpu_pose_task_centers=nullptr; // By Claude on 07/16/2026 (rung B1)
p->pose_task_capacity=0; p->pose_task_count=0; // By Claude on 07/16/2026 (rung B1) p->pose_task_capacity=0; p->pose_task_count=0; // By Claude on 07/16/2026 (rung B1)
p->bayer_staging[0]=p->bayer_staging[1]=nullptr; // By Claude on 07/16/2026 (rung B4)
p->bayer_copy_stream=nullptr; p->bayer_copy_event=nullptr; p->bayer_copy_pending=false;
return p; return p;
} }
...@@ -810,12 +825,58 @@ int tp_proc_set_kernels(TpProc* p, int cam, const float* d, int n){ if(!p||cam<0 ...@@ -810,12 +825,58 @@ int tp_proc_set_kernels(TpProc* p, int cam, const float* d, int n){ if(!p||cam<0
return cudaMemcpy(p->kernels_h[cam],d,(size_t)n*sizeof(float),cudaMemcpyHostToDevice)==cudaSuccess?0:-2; } return cudaMemcpy(p->kernels_h[cam],d,(size_t)n*sizeof(float),cudaMemcpyHostToDevice)==cudaSuccess?0:-2; }
int tp_proc_set_kernel_offsets(TpProc* p, int cam, const float* d, int n){ if(!p||cam<0||cam>=p->num_cams)return -1; cuCtxSetCurrent(p->mod->ctx); int tp_proc_set_kernel_offsets(TpProc* p, int cam, const float* d, int n){ if(!p||cam<0||cam>=p->num_cams)return -1; cuCtxSetCurrent(p->mod->ctx);
return cudaMemcpy(p->offs_h[cam],d,(size_t)n*sizeof(float),cudaMemcpyHostToDevice)==cudaSuccess?0:-2; } return cudaMemcpy(p->offs_h[cam],d,(size_t)n*sizeof(float),cudaMemcpyHostToDevice)==cudaSuccess?0:-2; }
// ---- 3-B rung B4: pinned staged bayer upload (async per-sensor H2D) ----
// Host-blocking fence on outstanding staged copies. The copy stream is
// NON-blocking (legacy-stream launches do NOT implicitly wait on it), so every
// consumer/writer of the resident images calls this first. Kernel exec paths
// (launch1 / exec_convert_direct) also cuCtxSynchronize before launching -
// the fence makes the direct-memcpy paths (get/set image) safe too.
// By Claude on 07/16/2026.
static void bayer_fence(TpProc* p){
if(p->bayer_copy_pending){ cudaEventSynchronize(p->bayer_copy_event); p->bayer_copy_pending=false; }
}
// Pinned host staging for slot (0/1): num_cams*img_w*img_h floats, lazily
// allocated (non-RT users never pay the pinned memory). Also lazily creates
// the copy stream + fence event. Returns NULL on failure (see tp_last_error).
float* tp_proc_bayer_staging(TpProc* p, int slot){
if(!p||slot<0||slot>1){ seterr("bayer_staging: bad args"); return nullptr; }
cuCtxSetCurrent(p->mod->ctx);
if(!p->bayer_staging[slot]){
size_t n=(size_t)p->num_cams*p->img_w*p->img_h*sizeof(float);
if(cudaHostAlloc((void**)&p->bayer_staging[slot], n, cudaHostAllocDefault)!=cudaSuccess){
p->bayer_staging[slot]=nullptr; seterr("bayer_staging: pinned alloc failed (%zu bytes)", n); return nullptr; }
}
if(!p->bayer_copy_stream && cudaStreamCreateWithFlags(&p->bayer_copy_stream, cudaStreamNonBlocking)!=cudaSuccess){
p->bayer_copy_stream=nullptr; seterr("bayer_staging: copy-stream create failed"); return nullptr; }
if(!p->bayer_copy_event && cudaEventCreateWithFlags(&p->bayer_copy_event, cudaEventDisableTiming)!=cudaSuccess){
p->bayer_copy_event=nullptr; seterr("bayer_staging: fence-event create failed"); return nullptr; }
return p->bayer_staging[slot];
}
// Enqueue the pinned->device copy of ONE sensor image from staging slot on the
// copy stream and move the fence event past it. Returns immediately - the host
// prepares the next sensor while this one DMAs. Caller contract: the sensor's
// region of the slot is fully written before this call and not rewritten until
// the next fence (per-scene slot flip keeps one full scene of slack).
int tp_proc_set_image_staged(TpProc* p, int cam, int slot){
if(!p||cam<0||cam>=p->num_cams||slot<0||slot>1) return -1;
if(!p->bayer_staging[slot]||!p->bayer_copy_stream||!p->bayer_copy_event){ seterr("set_image_staged: staging not ready (call tp_proc_bayer_staging)"); return -2; }
cuCtxSetCurrent(p->mod->ctx);
const float* src = p->bayer_staging[slot] + (size_t)cam*p->img_w*p->img_h;
if(cudaMemcpy2DAsync(p->images_h[cam], p->dstride_img, src, (size_t)p->img_w*sizeof(float),
(size_t)p->img_w*sizeof(float), p->img_h, cudaMemcpyHostToDevice,
p->bayer_copy_stream)!=cudaSuccess){ seterr("set_image_staged: async H2D failed cam=%d",cam); return -3; }
if(cudaEventRecord(p->bayer_copy_event, p->bayer_copy_stream)!=cudaSuccess){ seterr("set_image_staged: event record failed"); return -4; }
p->bayer_copy_pending=true;
return 0;
}
// pitched image upload (reuses harness update_image_gpu: dstride in floats) // pitched image upload (reuses harness update_image_gpu: dstride in floats)
int tp_proc_set_image(TpProc* p, int cam, const float* d){ if(!p||cam<0||cam>=p->num_cams)return -1; cuCtxSetCurrent(p->mod->ctx); int tp_proc_set_image(TpProc* p, int cam, const float* d){ if(!p||cam<0||cam>=p->num_cams)return -1; cuCtxSetCurrent(p->mod->ctx);
bayer_fence(p); // WAW order vs in-flight staged copies // By Claude on 07/16/2026 (rung B4)
update_image_gpu((float*)d, p->images_h[cam], p->dstride_img, p->img_w, p->img_h); return 0; } // dstride in BYTES (cudaMemcpy2D dpitch) update_image_gpu((float*)d, p->images_h[cam], p->dstride_img, p->img_w, p->img_h); return 0; } // dstride in BYTES (cudaMemcpy2D dpitch)
// De-pitch one resident source image. Primarily a correctness gate for // De-pitch one resident source image. Primarily a correctness gate for
// conditioning; also useful for explicit ingest diagnostics. // conditioning; also useful for explicit ingest diagnostics.
int tp_proc_get_image(TpProc* p, int cam, float* out){ if(!p||!out||cam<0||cam>=p->num_cams)return -1; cuCtxSetCurrent(p->mod->ctx); int tp_proc_get_image(TpProc* p, int cam, float* out){ if(!p||!out||cam<0||cam>=p->num_cams)return -1; cuCtxSetCurrent(p->mod->ctx);
bayer_fence(p); // By Claude on 07/16/2026 (rung B4)
return cudaMemcpy2D(out,(size_t)p->img_w*sizeof(float), p->images_h[cam],p->dstride_img, return cudaMemcpy2D(out,(size_t)p->img_w*sizeof(float), p->images_h[cam],p->dstride_img,
(size_t)p->img_w*sizeof(float),p->img_h,cudaMemcpyDeviceToHost)==cudaSuccess?0:-2; } (size_t)p->img_w*sizeof(float),p->img_h,cudaMemcpyDeviceToHost)==cudaSuccess?0:-2; }
...@@ -858,6 +919,7 @@ int tp_proc_exec_conditioning(TpProc* p, int use_double, double dc_level){ ...@@ -858,6 +919,7 @@ int tp_proc_exec_conditioning(TpProc* p, int use_double, double dc_level){
if(!p){ seterr("exec_conditioning: null"); return -1; } if(!p){ seterr("exec_conditioning: null"); return -1; }
if(use_double ? !p->cond_oracle_ready : !p->cond_maps_ready){ seterr("exec_conditioning: calibration not uploaded"); return -2; } if(use_double ? !p->cond_oracle_ready : !p->cond_maps_ready){ seterr("exec_conditioning: calibration not uploaded"); return -2; }
cuCtxSetCurrent(p->mod->ctx); cuCtxSetCurrent(p->mod->ctx);
bayer_fence(p); // staged uploads must land before in-place conditioning // By Claude on 07/16/2026 (rung B4)
const char* name=use_double?"condition_lwir_d":"condition_lwir_f"; const char* name=use_double?"condition_lwir_d":"condition_lwir_f";
CUfunction f=getfun(p->mod,name); if(!f){ seterr("%s missing",name); return -3; } CUfunction f=getfun(p->mod,name); if(!f){ seterr("%s missing",name); return -3; }
int nc=p->num_cams,w=p->img_w,h=p->img_h; size_t stride=p->dstride_img/sizeof(float); int nc=p->num_cams,w=p->img_w,h=p->img_h; size_t stride=p->dstride_img/sizeof(float);
...@@ -868,6 +930,7 @@ int tp_proc_exec_conditioning(TpProc* p, int use_double, double dc_level){ ...@@ -868,6 +930,7 @@ int tp_proc_exec_conditioning(TpProc* p, int use_double, double dc_level){
} }
// use_center_image: broadcast one center image to all sensors (FPN back-prop mode) // use_center_image: broadcast one center image to all sensors (FPN back-prop mode)
int tp_proc_set_center_image(TpProc* p, const float* d){ if(!p)return -1; cuCtxSetCurrent(p->mod->ctx); int tp_proc_set_center_image(TpProc* p, const float* d){ if(!p)return -1; cuCtxSetCurrent(p->mod->ctx);
bayer_fence(p); // By Claude on 07/16/2026 (rung B4)
for(int c=0;c<p->num_cams;c++) update_image_gpu((float*)d, p->images_h[c], p->dstride_img, p->img_w, p->img_h); return 0; } for(int c=0;c<p->num_cams;c++) update_image_gpu((float*)d, p->images_h[c], p->dstride_img, p->img_w, p->img_h); return 0; }
int tp_proc_set_tasks_slot(TpProc* p, const float* ftasks, int ntiles, int total_floats, int slot){ int tp_proc_set_tasks_slot(TpProc* p, const float* ftasks, int ntiles, int total_floats, int slot){
if(!p||!ftasks||(slot<0)||(slot>1)||(ntiles<0)||(total_floats<=0)) return -1; if(!p||!ftasks||(slot<0)||(slot>1)||(ntiles<0)||(total_floats<=0)) return -1;
...@@ -947,6 +1010,7 @@ int tp_proc_exec_geometry(TpProc* p, int uniform_grid){ ...@@ -947,6 +1010,7 @@ int tp_proc_exec_geometry(TpProc* p, int uniform_grid){
// set time; ref_scene: write to clt_ref; erase_clt: -1 none / 0 zero / 1 NaN (erase_clt_tiles first). // set time; ref_scene: write to clt_ref; erase_clt: -1 none / 0 zero / 1 NaN (erase_clt_tiles first).
int tp_proc_exec_convert_direct(TpProc* p, int ref_scene, int erase_clt, int no_kernels){ int tp_proc_exec_convert_direct(TpProc* p, int ref_scene, int erase_clt, int no_kernels){
if(!p)return -1; cuCtxSetCurrent(p->mod->ctx); CUresult cr; const char* es; if(!p)return -1; cuCtxSetCurrent(p->mod->ctx); CUresult cr; const char* es;
bayer_fence(p); // staged uploads must land before convert reads images // By Claude on 07/16/2026 (rung B4)
CUfunction f_cd=getfun(p->mod,"convert_direct"), f_er=getfun(p->mod,"erase_clt_tiles"); CUfunction f_cd=getfun(p->mod,"convert_direct"), f_er=getfun(p->mod,"erase_clt_tiles");
if(!f_cd){ seterr("convert_direct missing"); return -2; } if(!f_cd){ seterr("convert_direct missing"); return -2; }
float** clt_sel = ref_scene ? p->gpu_clt_ref : p->gpu_clt; float** clt_sel = ref_scene ? p->gpu_clt_ref : p->gpu_clt;
...@@ -2313,6 +2377,11 @@ int tp_proc_convert_selftest(TpModule* m, int lwir, const char* data_root, ...@@ -2313,6 +2377,11 @@ int tp_proc_convert_selftest(TpModule* m, int lwir, const char* data_root,
void tp_proc_destroy(TpProc* p){ void tp_proc_destroy(TpProc* p){
if(!p)return; if(p->mod) cuCtxSetCurrent(p->mod->ctx); if(!p)return; if(p->mod) cuCtxSetCurrent(p->mod->ctx);
// rung B4: drain the copy stream before freeing pinned staging // By Claude on 07/16/2026
if(p->bayer_copy_stream){ cudaStreamSynchronize(p->bayer_copy_stream); cudaStreamDestroy(p->bayer_copy_stream); }
if(p->bayer_copy_event) cudaEventDestroy(p->bayer_copy_event);
if(p->bayer_staging[0]) cudaFreeHost(p->bayer_staging[0]);
if(p->bayer_staging[1]) cudaFreeHost(p->bayer_staging[1]);
for(int c=0;c<p->num_cams;c++){ if(c<(int)p->kernels_h.size())cudaFree(p->kernels_h[c]); if(c<(int)p->offs_h.size())cudaFree(p->offs_h[c]); for(int c=0;c<p->num_cams;c++){ if(c<(int)p->kernels_h.size())cudaFree(p->kernels_h[c]); if(c<(int)p->offs_h.size())cudaFree(p->offs_h[c]);
if(c<(int)p->images_h.size())cudaFree(p->images_h[c]); if(c<(int)p->clt_h.size())cudaFree(p->clt_h[c]); if(c<(int)p->clt_ref_h.size())cudaFree(p->clt_ref_h[c]); } if(c<(int)p->images_h.size())cudaFree(p->images_h[c]); if(c<(int)p->clt_h.size())cudaFree(p->clt_h[c]); if(c<(int)p->clt_ref_h.size())cudaFree(p->clt_ref_h[c]); }
cudaFree(p->gpu_kernels); cudaFree(p->gpu_kernel_offsets); cudaFree(p->gpu_images); cudaFree(p->gpu_clt); cudaFree(p->gpu_clt_ref); cudaFree(p->gpu_kernels); cudaFree(p->gpu_kernel_offsets); cudaFree(p->gpu_images); cudaFree(p->gpu_clt); cudaFree(p->gpu_clt_ref);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment