core: writer-owned relocation
The worker no longer copies rows when an array cannot grow in place. It reserves a zeroed region, returns the staged descriptor with a commit flag, and queues every other request until the main thread (the only writer) has copied the old bytes and sent commit-rows. Writes that raced the old worker-side copy were lost; lights in the glTF viewer regularly ended up at the origin because of it. Amp-Thread-ID: https://ampcode.com/threads/T-01a0b2dd-db32-712d-99ce-12ce4a86d979 Co-authored-by: Heaust Azure <heaust.azure@gmail.com>
This commit is contained in:
+34
-14
@@ -75,23 +75,17 @@ export class YawnCore {
|
|||||||
|
|
||||||
async createRows({ name, rows, stride, format }) {
|
async createRows({ name, rows, stride, format }) {
|
||||||
await this.ready;
|
await this.ready;
|
||||||
const descriptor = await this.#request("create-rows", { name, rows, stride, format });
|
const result = await this.#request("create-rows", { name, rows, stride, format });
|
||||||
const array = this.#arrays.get(name)?.update(descriptor)
|
return this.#adopt(result.rows, result.commit ? [name] : []);
|
||||||
?? new SharedRows(this.#buffer, descriptor);
|
|
||||||
this.#arrays.set(name, array);
|
|
||||||
return array;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async createRowsBatch(rows) {
|
async createRowsBatch(rows) {
|
||||||
await this.ready;
|
await this.ready;
|
||||||
if (!Array.isArray(rows) || !rows.length) throw new TypeError("ROWS");
|
if (!Array.isArray(rows) || !rows.length) throw new TypeError("ROWS");
|
||||||
const descriptors = await this.#request("create-rows-batch", { rows });
|
const result = await this.#request("create-rows-batch", { rows });
|
||||||
return descriptors.map(descriptor => {
|
const arrays = result.rows.map(descriptor => this.#adopt(descriptor));
|
||||||
const array = this.#arrays.get(descriptor.name)?.update(descriptor)
|
this.#commit(result.commit);
|
||||||
?? new SharedRows(this.#buffer, descriptor);
|
return arrays;
|
||||||
this.#arrays.set(descriptor.name, array);
|
|
||||||
return array;
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async deleteRows(name) {
|
async deleteRows(name) {
|
||||||
@@ -103,11 +97,37 @@ export class YawnCore {
|
|||||||
/** Reserves `count` contiguous rows and resolves with the first row ID. */
|
/** Reserves `count` contiguous rows and resolves with the first row ID. */
|
||||||
async allocateObject(name, count = 1) {
|
async allocateObject(name, count = 1) {
|
||||||
await this.ready;
|
await this.ready;
|
||||||
const { id, rows } = await this.#request("allocate-object", { name, count });
|
const { id, rows, commit } = await this.#request("allocate-object", { name, count });
|
||||||
this.#arrays.get(name).update(rows);
|
this.#adopt(rows, commit ? [name] : []);
|
||||||
return id;
|
return id;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Publishes a descriptor the worker returned. When the rows moved, this thread owns the copy:
|
||||||
|
* the old bytes are copied synchronously into the new region and the worker is told to commit
|
||||||
|
* before anything else can be requested, so no write to the array is ever lost.
|
||||||
|
*/
|
||||||
|
#adopt(descriptor, commit = []) {
|
||||||
|
const current = this.#arrays.get(descriptor.name);
|
||||||
|
if (!current) {
|
||||||
|
const array = new SharedRows(this.#buffer, descriptor);
|
||||||
|
this.#arrays.set(descriptor.name, array);
|
||||||
|
return array;
|
||||||
|
}
|
||||||
|
if (descriptor.offset !== current.descriptor.offset) {
|
||||||
|
const { offset, bytes } = current.descriptor;
|
||||||
|
new Uint8Array(this.#buffer, descriptor.offset, bytes).set(new Uint8Array(this.#buffer, offset, bytes));
|
||||||
|
}
|
||||||
|
current.update(descriptor);
|
||||||
|
this.#commit(commit);
|
||||||
|
return current;
|
||||||
|
}
|
||||||
|
|
||||||
|
#commit(names) {
|
||||||
|
if (!names.length) return;
|
||||||
|
this.#request("commit-rows", { names }).catch(error => this.#fail(error.code));
|
||||||
|
}
|
||||||
|
|
||||||
async deleteObject(name, id) {
|
async deleteObject(name, id) {
|
||||||
await this.ready;
|
await this.ready;
|
||||||
return this.#request("delete-object", { name, id });
|
return this.#request("delete-object", { name, id });
|
||||||
|
|||||||
+61
-24
@@ -93,6 +93,8 @@ impl Core {
|
|||||||
serde_json::to_string(&gpu.as_ref().map(|gpu| &gpu.features[..]).unwrap_or(&[])).unwrap()
|
serde_json::to_string(&gpu.as_ref().map(|gpu| &gpu.features[..]).unwrap_or(&[])).unwrap()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Creates or grows an array. When `commit` is set the rows moved: the caller must copy the old
|
||||||
|
/// bytes into the returned descriptor and send `commit_rows` before any other request.
|
||||||
pub fn create_rows(
|
pub fn create_rows(
|
||||||
&self,
|
&self,
|
||||||
name: String,
|
name: String,
|
||||||
@@ -100,20 +102,18 @@ impl Core {
|
|||||||
stride: u32,
|
stride: u32,
|
||||||
format: String,
|
format: String,
|
||||||
) -> Result<String, JsError> {
|
) -> Result<String, JsError> {
|
||||||
let rows = self
|
let (rows, moved) = self
|
||||||
.data
|
.data
|
||||||
.borrow_mut()
|
.borrow_mut()
|
||||||
.create_rows(name, rows, stride, format)
|
.create_rows(name, rows, stride, format)
|
||||||
.map_err(JsError::new)?;
|
.map_err(JsError::new)?;
|
||||||
if let Some(gpu) = self.gpu.borrow().as_ref() {
|
if !moved {
|
||||||
self.store
|
self.refresh_rows(&rows.name)?;
|
||||||
.borrow_mut()
|
|
||||||
.refresh_rows(&rows.name, gpu, &self.data.borrow())
|
|
||||||
.map_err(|error| JsError::new(&error))?;
|
|
||||||
}
|
}
|
||||||
Ok(serde_json::to_string(&rows).unwrap())
|
Ok(serde_json::json!({ "rows": rows, "commit": moved }).to_string())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Like `create_rows` for several arrays; `commit` lists the arrays that moved.
|
||||||
pub fn create_rows_batch(&self, source: &str) -> Result<String, JsError> {
|
pub fn create_rows_batch(&self, source: &str) -> Result<String, JsError> {
|
||||||
let requests: Vec<RowRequest> =
|
let requests: Vec<RowRequest> =
|
||||||
serde_json::from_str(source).map_err(|_| JsError::new("ROWS"))?;
|
serde_json::from_str(source).map_err(|_| JsError::new("ROWS"))?;
|
||||||
@@ -123,25 +123,66 @@ impl Core {
|
|||||||
let refresh = requests
|
let refresh = requests
|
||||||
.iter()
|
.iter()
|
||||||
.any(|request| self.store.borrow().uses_rows(&request.name));
|
.any(|request| self.store.borrow().uses_rows(&request.name));
|
||||||
let descriptors = {
|
let results = {
|
||||||
let mut data = self.data.borrow_mut();
|
let mut data = self.data.borrow_mut();
|
||||||
requests
|
let results = requests
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|request| {
|
.map(|request| {
|
||||||
data.create_rows(request.name, request.rows, request.stride, request.format)
|
data.create_rows(request.name, request.rows, request.stride, request.format)
|
||||||
.map_err(JsError::new)
|
|
||||||
})
|
})
|
||||||
.collect::<Result<Vec<_>, _>>()?
|
.collect::<Result<Vec<_>, _>>();
|
||||||
|
if results.is_err() {
|
||||||
|
data.discard_pending();
|
||||||
|
}
|
||||||
|
results.map_err(JsError::new)?
|
||||||
};
|
};
|
||||||
if refresh {
|
let commit = results
|
||||||
|
.iter()
|
||||||
|
.filter(|(_, moved)| *moved)
|
||||||
|
.map(|(rows, _)| rows.name.clone())
|
||||||
|
.collect::<Vec<_>>();
|
||||||
|
if refresh && results.iter().any(|(_, moved)| !moved) {
|
||||||
|
self.refresh()?;
|
||||||
|
}
|
||||||
|
let descriptors = results
|
||||||
|
.into_iter()
|
||||||
|
.map(|(rows, _)| rows)
|
||||||
|
.collect::<Vec<_>>();
|
||||||
|
Ok(serde_json::json!({ "rows": descriptors, "commit": commit }).to_string())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Publishes relocations the caller finished copying, then rebuilds the GPU side once.
|
||||||
|
pub fn commit_rows(&self, source: &str) -> Result<(), JsError> {
|
||||||
|
let names: Vec<String> = serde_json::from_str(source).map_err(|_| JsError::new("ROWS"))?;
|
||||||
|
self.data
|
||||||
|
.borrow_mut()
|
||||||
|
.commit(&names)
|
||||||
|
.map_err(JsError::new)?;
|
||||||
|
if names.iter().any(|name| self.store.borrow().uses_rows(name)) {
|
||||||
|
self.refresh()?;
|
||||||
|
}
|
||||||
|
self.data.borrow_mut().mark_dirty();
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn refresh_rows(&self, name: &str) -> Result<(), JsError> {
|
||||||
|
if let Some(gpu) = self.gpu.borrow().as_ref() {
|
||||||
|
self.store
|
||||||
|
.borrow_mut()
|
||||||
|
.refresh_rows(name, gpu, &self.data.borrow())
|
||||||
|
.map_err(|error| JsError::new(&error))?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn refresh(&self) -> Result<(), JsError> {
|
||||||
if let Some(gpu) = self.gpu.borrow().as_ref() {
|
if let Some(gpu) = self.gpu.borrow().as_ref() {
|
||||||
self.store
|
self.store
|
||||||
.borrow_mut()
|
.borrow_mut()
|
||||||
.refresh(gpu, &self.data.borrow())
|
.refresh(gpu, &self.data.borrow())
|
||||||
.map_err(|error| JsError::new(&error))?;
|
.map_err(|error| JsError::new(&error))?;
|
||||||
}
|
}
|
||||||
}
|
Ok(())
|
||||||
Ok(serde_json::to_string(&descriptors).unwrap())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn delete_rows(&self, name: String) -> Result<(), JsError> {
|
pub fn delete_rows(&self, name: String) -> Result<(), JsError> {
|
||||||
@@ -154,8 +195,9 @@ impl Core {
|
|||||||
.map_err(JsError::new)
|
.map_err(JsError::new)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Reserves rows; `commit` is set when the array moved, exactly like `create_rows`.
|
||||||
pub fn allocate_object(&self, name: &str, count: u32) -> Result<String, JsError> {
|
pub fn allocate_object(&self, name: &str, count: u32) -> Result<String, JsError> {
|
||||||
let (id, grew) = self
|
let (id, grew, moved) = self
|
||||||
.data
|
.data
|
||||||
.borrow_mut()
|
.borrow_mut()
|
||||||
.allocate_object(name, count)
|
.allocate_object(name, count)
|
||||||
@@ -163,18 +205,13 @@ impl Core {
|
|||||||
let rows = self
|
let rows = self
|
||||||
.data
|
.data
|
||||||
.borrow()
|
.borrow()
|
||||||
.rows(name)
|
.staged(name)
|
||||||
.ok_or_else(|| JsError::new("ROWS_UNKNOWN"))?
|
.ok_or_else(|| JsError::new("ROWS_UNKNOWN"))?
|
||||||
.clone();
|
.clone();
|
||||||
if grew {
|
if grew && !moved {
|
||||||
if let Some(gpu) = self.gpu.borrow().as_ref() {
|
self.refresh_rows(name)?;
|
||||||
self.store
|
|
||||||
.borrow_mut()
|
|
||||||
.refresh_rows(name, gpu, &self.data.borrow())
|
|
||||||
.map_err(|error| JsError::new(&error))?;
|
|
||||||
}
|
}
|
||||||
}
|
Ok(serde_json::json!({ "id": id, "rows": rows, "commit": moved }).to_string())
|
||||||
Ok(serde_json::json!({ "id": id, "rows": rows }).to_string())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn delete_object(&self, name: &str, id: u32) -> Result<(), JsError> {
|
pub fn delete_object(&self, name: &str, id: u32) -> Result<(), JsError> {
|
||||||
|
|||||||
+78
-27
@@ -27,15 +27,19 @@ impl FreeList {
|
|||||||
|
|
||||||
/// Claims exactly `[offset, offset + size)` when that range is entirely free.
|
/// Claims exactly `[offset, offset + size)` when that range is entirely free.
|
||||||
fn take(&mut self, offset: u32, size: u32) -> bool {
|
fn take(&mut self, offset: u32, size: u32) -> bool {
|
||||||
let Some(&available) = self.0.get(&offset) else {
|
let Some((&block, &block_size)) = self.0.range(..=offset).next_back() else {
|
||||||
return false;
|
return false;
|
||||||
};
|
};
|
||||||
if available < size {
|
if offset + size > block + block_size {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
self.0.remove(&offset);
|
self.0.remove(&block);
|
||||||
if available > size {
|
if offset > block {
|
||||||
self.0.insert(offset + size, available - size);
|
self.0.insert(block, offset - block);
|
||||||
|
}
|
||||||
|
let end = offset + size;
|
||||||
|
if end < block + block_size {
|
||||||
|
self.0.insert(end, block + block_size - end);
|
||||||
}
|
}
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
@@ -58,7 +62,7 @@ impl FreeList {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Serialize)]
|
#[derive(Clone, Debug, PartialEq, Serialize)]
|
||||||
pub struct Rows {
|
pub struct Rows {
|
||||||
pub name: String,
|
pub name: String,
|
||||||
pub rows: u32,
|
pub rows: u32,
|
||||||
@@ -82,6 +86,8 @@ pub struct RenderData {
|
|||||||
free: FreeList,
|
free: FreeList,
|
||||||
rows: HashMap<String, Rows>,
|
rows: HashMap<String, Rows>,
|
||||||
slots: HashMap<String, Slots>,
|
slots: HashMap<String, Slots>,
|
||||||
|
/// Relocated descriptors the writer has not committed yet; `rows` keeps the old storage live.
|
||||||
|
pending: HashMap<String, Rows>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RenderData {
|
impl RenderData {
|
||||||
@@ -99,19 +105,22 @@ impl RenderData {
|
|||||||
free: FreeList(BTreeMap::from([(0, capacity)])),
|
free: FreeList(BTreeMap::from([(0, capacity)])),
|
||||||
rows: HashMap::new(),
|
rows: HashMap::new(),
|
||||||
slots: HashMap::new(),
|
slots: HashMap::new(),
|
||||||
|
pending: HashMap::new(),
|
||||||
};
|
};
|
||||||
data.create_rows("signals".into(), 1, 32, "f32".into())?;
|
data.create_rows("signals".into(), 1, 32, "f32".into())?;
|
||||||
data.write_signal(5, 1.0);
|
data.write_signal(5, 1.0);
|
||||||
Ok(data)
|
Ok(data)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Creates or grows an array. The flag is set when the rows moved: the returned descriptor
|
||||||
|
/// is then staged and the writer must copy the old bytes into it and `commit` it.
|
||||||
pub fn create_rows(
|
pub fn create_rows(
|
||||||
&mut self,
|
&mut self,
|
||||||
name: String,
|
name: String,
|
||||||
rows: u32,
|
rows: u32,
|
||||||
stride: u32,
|
stride: u32,
|
||||||
format: String,
|
format: String,
|
||||||
) -> Result<Rows, &'static str> {
|
) -> Result<(Rows, bool), &'static str> {
|
||||||
if name.is_empty()
|
if name.is_empty()
|
||||||
|| rows == 0
|
|| rows == 0
|
||||||
|| stride < 16
|
|| stride < 16
|
||||||
@@ -125,19 +134,19 @@ impl RenderData {
|
|||||||
return Err("ROWS");
|
return Err("ROWS");
|
||||||
}
|
}
|
||||||
if rows <= current.rows {
|
if rows <= current.rows {
|
||||||
return Ok(current);
|
return Ok((current, false));
|
||||||
}
|
}
|
||||||
if name == "signals" {
|
if name == "signals" {
|
||||||
return Err("ROWS_BUILTIN");
|
return Err("ROWS_BUILTIN");
|
||||||
}
|
}
|
||||||
let added = rows - current.rows;
|
let added = rows - current.rows;
|
||||||
let descriptor = self.grow_rows(current, rows)?;
|
let (descriptor, moved) = self.grow_rows(current, rows)?;
|
||||||
self.slots
|
self.slots
|
||||||
.get_mut(&descriptor.name)
|
.get_mut(&descriptor.name)
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.free
|
.free
|
||||||
.release(descriptor.rows - added, added);
|
.release(descriptor.rows - added, added);
|
||||||
return Ok(descriptor);
|
return Ok((descriptor, moved));
|
||||||
}
|
}
|
||||||
let bytes = rows.checked_mul(stride).ok_or("ARENA_OOM")?;
|
let bytes = rows.checked_mul(stride).ok_or("ARENA_OOM")?;
|
||||||
let offset = self.reserve(bytes)?;
|
let offset = self.reserve(bytes)?;
|
||||||
@@ -155,13 +164,16 @@ impl RenderData {
|
|||||||
let mut slots = Slots::default();
|
let mut slots = Slots::default();
|
||||||
slots.free.release(0, rows);
|
slots.free.release(0, rows);
|
||||||
self.slots.insert(descriptor.name.clone(), slots);
|
self.slots.insert(descriptor.name.clone(), slots);
|
||||||
Ok(descriptor)
|
Ok((descriptor, false))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn delete_rows(&mut self, name: &str) -> Result<(), &'static str> {
|
pub fn delete_rows(&mut self, name: &str) -> Result<(), &'static str> {
|
||||||
if name == "signals" {
|
if name == "signals" {
|
||||||
return Err("ROWS_BUILTIN");
|
return Err("ROWS_BUILTIN");
|
||||||
}
|
}
|
||||||
|
if self.pending.contains_key(name) {
|
||||||
|
return Err("ROWS_PENDING");
|
||||||
|
}
|
||||||
if self
|
if self
|
||||||
.slots
|
.slots
|
||||||
.get(name)
|
.get(name)
|
||||||
@@ -192,7 +204,12 @@ impl RenderData {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Reserves `count` contiguous rows and returns the first one; grows geometrically when full.
|
/// Reserves `count` contiguous rows and returns the first one; grows geometrically when full.
|
||||||
pub fn allocate_object(&mut self, name: &str, count: u32) -> Result<(u32, bool), &'static str> {
|
/// The growth flags mirror `create_rows`: `(grew, moved)`.
|
||||||
|
pub fn allocate_object(
|
||||||
|
&mut self,
|
||||||
|
name: &str,
|
||||||
|
count: u32,
|
||||||
|
) -> Result<(u32, bool, bool), &'static str> {
|
||||||
if name == "signals" {
|
if name == "signals" {
|
||||||
return Err("ROWS_BUILTIN");
|
return Err("ROWS_BUILTIN");
|
||||||
}
|
}
|
||||||
@@ -201,6 +218,7 @@ impl RenderData {
|
|||||||
}
|
}
|
||||||
let slots = self.slots.get_mut(name).ok_or("ROWS_UNKNOWN")?;
|
let slots = self.slots.get_mut(name).ok_or("ROWS_UNKNOWN")?;
|
||||||
let mut grew = false;
|
let mut grew = false;
|
||||||
|
let mut moved = false;
|
||||||
let id = match slots.free.allocate(count, 1) {
|
let id = match slots.free.allocate(count, 1) {
|
||||||
Some(id) => id,
|
Some(id) => id,
|
||||||
None => {
|
None => {
|
||||||
@@ -211,7 +229,7 @@ impl RenderData {
|
|||||||
.ok_or("OBJECT_LIMIT")?
|
.ok_or("OBJECT_LIMIT")?
|
||||||
.max(descriptor.rows.checked_add(count).ok_or("OBJECT_LIMIT")?);
|
.max(descriptor.rows.checked_add(count).ok_or("OBJECT_LIMIT")?);
|
||||||
let old_rows = descriptor.rows;
|
let old_rows = descriptor.rows;
|
||||||
self.grow_rows(descriptor, rows)?;
|
moved = self.grow_rows(descriptor, rows)?.1;
|
||||||
grew = true;
|
grew = true;
|
||||||
let slots = self.slots.get_mut(name).unwrap();
|
let slots = self.slots.get_mut(name).unwrap();
|
||||||
slots.free.release(old_rows, rows - old_rows);
|
slots.free.release(old_rows, rows - old_rows);
|
||||||
@@ -219,7 +237,35 @@ impl RenderData {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
self.slots.get_mut(name).unwrap().spans.insert(id, count);
|
self.slots.get_mut(name).unwrap().spans.insert(id, count);
|
||||||
Ok((id, grew))
|
Ok((id, grew, moved))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The descriptor the writer should use: staged rows win over committed ones.
|
||||||
|
pub fn staged(&self, name: &str) -> Option<&Rows> {
|
||||||
|
self.pending.get(name).or_else(|| self.rows.get(name))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Publishes staged relocations after the writer copied the old bytes into them.
|
||||||
|
pub fn commit(&mut self, names: &[String]) -> Result<(), &'static str> {
|
||||||
|
for name in names {
|
||||||
|
let staged = self.pending.remove(name).ok_or("ROWS_PENDING")?;
|
||||||
|
let old = self.rows.insert(name.clone(), staged).unwrap();
|
||||||
|
self.release(old.offset - self.base, old.bytes);
|
||||||
|
}
|
||||||
|
if !self.pending.is_empty() {
|
||||||
|
return Err("ROWS_PENDING");
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drops every staged relocation, for example when a batch fails half-way.
|
||||||
|
pub fn discard_pending(&mut self) {
|
||||||
|
for staged in std::mem::take(&mut self.pending).into_values() {
|
||||||
|
self.release(staged.offset - self.base, staged.bytes);
|
||||||
|
let rows = self.rows[&staged.name].rows;
|
||||||
|
let slots = self.slots.get_mut(&staged.name).unwrap();
|
||||||
|
assert!(slots.free.take(rows, staged.rows - rows));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn delete_object(&mut self, name: &str, id: u32) -> Result<(), &'static str> {
|
pub fn delete_object(&mut self, name: &str, id: u32) -> Result<(), &'static str> {
|
||||||
@@ -232,31 +278,36 @@ impl RenderData {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn grow_rows(&mut self, mut descriptor: Rows, rows: u32) -> Result<Rows, &'static str> {
|
/// Extends in place when the next arena block is free; otherwise stages a zeroed region for
|
||||||
|
/// the writer to fill and `commit`, because only the writer knows when its last write landed.
|
||||||
|
fn grow_rows(&mut self, mut descriptor: Rows, rows: u32) -> Result<(Rows, bool), &'static str> {
|
||||||
|
if self.pending.contains_key(&descriptor.name) {
|
||||||
|
return Err("ROWS_PENDING");
|
||||||
|
}
|
||||||
let bytes = rows.checked_mul(descriptor.stride).ok_or("ARENA_OOM")?;
|
let bytes = rows.checked_mul(descriptor.stride).ok_or("ARENA_OOM")?;
|
||||||
let old_offset = descriptor.offset - self.base;
|
let old_offset = descriptor.offset - self.base;
|
||||||
let old_end = old_offset + descriptor.bytes;
|
let old_end = old_offset + descriptor.bytes;
|
||||||
let extra = bytes - descriptor.bytes;
|
let extra = bytes - descriptor.bytes;
|
||||||
if self.free.take(old_end, extra) {
|
let moved = !self.free.take(old_end, extra);
|
||||||
self.arena[self.arena_start + old_end as usize
|
if moved {
|
||||||
..self.arena_start + (old_end + extra) as usize]
|
|
||||||
.fill(0);
|
|
||||||
} else {
|
|
||||||
let offset = self.reserve(bytes)?;
|
let offset = self.reserve(bytes)?;
|
||||||
let start = self.arena_start + offset as usize;
|
let start = self.arena_start + offset as usize;
|
||||||
self.arena[start..start + bytes as usize].fill(0);
|
self.arena[start..start + bytes as usize].fill(0);
|
||||||
self.arena.copy_within(
|
|
||||||
self.arena_start + old_offset as usize..self.arena_start + old_end as usize,
|
|
||||||
start,
|
|
||||||
);
|
|
||||||
self.release(old_offset, descriptor.bytes);
|
|
||||||
descriptor.offset = self.base + offset;
|
descriptor.offset = self.base + offset;
|
||||||
|
} else {
|
||||||
|
self.arena[self.arena_start + old_end as usize
|
||||||
|
..self.arena_start + (old_end + extra) as usize]
|
||||||
|
.fill(0);
|
||||||
}
|
}
|
||||||
descriptor.rows = rows;
|
descriptor.rows = rows;
|
||||||
descriptor.bytes = bytes;
|
descriptor.bytes = bytes;
|
||||||
self.rows
|
if moved {
|
||||||
|
&mut self.pending
|
||||||
|
} else {
|
||||||
|
&mut self.rows
|
||||||
|
}
|
||||||
.insert(descriptor.name.clone(), descriptor.clone());
|
.insert(descriptor.name.clone(), descriptor.clone());
|
||||||
Ok(descriptor)
|
Ok((descriptor, moved))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn update_signals(&mut self, delta: f32, frame: u32, elapsed: f32, fps: u32) {
|
pub fn update_signals(&mut self, delta: f32, frame: u32, elapsed: f32, fps: u32) {
|
||||||
|
|||||||
+68
-19
@@ -40,22 +40,23 @@ fn spans_reuse_coalesced_gaps_before_growing() {
|
|||||||
let mut data = RenderData::new(4096).unwrap();
|
let mut data = RenderData::new(4096).unwrap();
|
||||||
data.create_rows("v".into(), 8, 16, "f32".into()).unwrap();
|
data.create_rows("v".into(), 8, 16, "f32".into()).unwrap();
|
||||||
|
|
||||||
assert_eq!(data.allocate_object("v", 2).unwrap(), (0, false));
|
assert_eq!(data.allocate_object("v", 2).unwrap(), (0, false, false));
|
||||||
assert_eq!(data.allocate_object("v", 3).unwrap(), (2, false));
|
assert_eq!(data.allocate_object("v", 3).unwrap(), (2, false, false));
|
||||||
assert_eq!(data.allocate_object("v", 3).unwrap(), (5, false));
|
assert_eq!(data.allocate_object("v", 3).unwrap(), (5, false, false));
|
||||||
assert_eq!(rows_of(&data, "v"), 8);
|
assert_eq!(rows_of(&data, "v"), 8);
|
||||||
|
|
||||||
// Freeing the two middle spans leaves one coalesced hole of 6 rows.
|
// Freeing the two middle spans leaves one coalesced hole of 6 rows.
|
||||||
data.delete_object("v", 2).unwrap();
|
data.delete_object("v", 2).unwrap();
|
||||||
data.delete_object("v", 5).unwrap();
|
data.delete_object("v", 5).unwrap();
|
||||||
assert_eq!(data.delete_object("v", 5), Err("OBJECT_UNKNOWN"));
|
assert_eq!(data.delete_object("v", 5), Err("OBJECT_UNKNOWN"));
|
||||||
assert_eq!(data.allocate_object("v", 6).unwrap(), (2, false));
|
assert_eq!(data.allocate_object("v", 6).unwrap(), (2, false, false));
|
||||||
assert_eq!(rows_of(&data, "v"), 8);
|
assert_eq!(rows_of(&data, "v"), 8);
|
||||||
|
|
||||||
// A span that does not fit doubles capacity instead of growing to the exact need.
|
// A span that does not fit doubles capacity instead of growing to the exact need; the only
|
||||||
assert_eq!(data.allocate_object("v", 1).unwrap(), (8, true));
|
// array in the arena extends in place, so nothing is staged.
|
||||||
|
assert_eq!(data.allocate_object("v", 1).unwrap(), (8, true, false));
|
||||||
assert_eq!(rows_of(&data, "v"), 16);
|
assert_eq!(rows_of(&data, "v"), 16);
|
||||||
assert_eq!(data.allocate_object("v", 40).unwrap(), (9, true));
|
assert_eq!(data.allocate_object("v", 40).unwrap(), (9, true, false));
|
||||||
assert_eq!(rows_of(&data, "v"), 56);
|
assert_eq!(rows_of(&data, "v"), 56);
|
||||||
assert_eq!(data.allocate_object("v", 0), Err("ROWS"));
|
assert_eq!(data.allocate_object("v", 0), Err("ROWS"));
|
||||||
}
|
}
|
||||||
@@ -64,8 +65,8 @@ fn spans_reuse_coalesced_gaps_before_growing() {
|
|||||||
fn deleting_a_span_zeroes_every_row_and_keeps_neighbours() {
|
fn deleting_a_span_zeroes_every_row_and_keeps_neighbours() {
|
||||||
let mut data = RenderData::new(4096).unwrap();
|
let mut data = RenderData::new(4096).unwrap();
|
||||||
data.create_rows("v".into(), 4, 16, "u32".into()).unwrap();
|
data.create_rows("v".into(), 4, 16, "u32".into()).unwrap();
|
||||||
let (first, _) = data.allocate_object("v", 2).unwrap();
|
let (first, ..) = data.allocate_object("v", 2).unwrap();
|
||||||
let (second, _) = data.allocate_object("v", 1).unwrap();
|
let (second, ..) = data.allocate_object("v", 1).unwrap();
|
||||||
let base = (data.rows("v").unwrap().offset - data.base) as usize + data.arena_start;
|
let base = (data.rows("v").unwrap().offset - data.base) as usize + data.arena_start;
|
||||||
data.arena[base..base + 48].fill(0xAB);
|
data.arena[base..base + 48].fill(0xAB);
|
||||||
|
|
||||||
@@ -80,20 +81,68 @@ fn deleting_a_span_zeroes_every_row_and_keeps_neighbours() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn explicit_growth_feeds_the_span_allocator_and_relocation_keeps_content() {
|
fn relocation_is_staged_until_the_writer_commits() {
|
||||||
let mut data = RenderData::new(4096).unwrap();
|
let mut data = RenderData::new(4096).unwrap();
|
||||||
data.create_rows("a".into(), 1, 16, "u32".into()).unwrap();
|
data.create_rows("a".into(), 1, 16, "u32".into()).unwrap();
|
||||||
data.create_rows("b".into(), 1, 16, "u32".into()).unwrap();
|
data.create_rows("b".into(), 1, 16, "u32".into()).unwrap();
|
||||||
assert_eq!(data.allocate_object("a", 1).unwrap(), (0, false));
|
assert_eq!(data.allocate_object("a", 1).unwrap(), (0, false, false));
|
||||||
let base = (data.rows("a").unwrap().offset - data.base) as usize + data.arena_start;
|
let old = data.rows("a").unwrap().clone();
|
||||||
|
let base = (old.offset - data.base) as usize + data.arena_start;
|
||||||
data.arena[base..base + 16].fill(0xCD);
|
data.arena[base..base + 16].fill(0xCD);
|
||||||
|
|
||||||
let grown = data.create_rows("a".into(), 5, 16, "u32".into()).unwrap();
|
// "b" sits right behind "a", so growing "a" must move it: the worker stages a zeroed region
|
||||||
assert_eq!(grown.rows, 5);
|
// and keeps rendering from (and the arena owning) the old bytes.
|
||||||
assert!(data.bytes("a").unwrap()[..16]
|
let (staged, moved) = data.create_rows("a".into(), 5, 16, "u32".into()).unwrap();
|
||||||
.iter()
|
assert!(moved);
|
||||||
.all(|byte| *byte == 0xCD));
|
assert_eq!(staged.rows, 5);
|
||||||
|
assert_ne!(staged.offset, old.offset);
|
||||||
|
assert_eq!(data.rows("a").unwrap().offset, old.offset);
|
||||||
|
assert_eq!(data.staged("a").unwrap().offset, staged.offset);
|
||||||
|
assert_eq!(
|
||||||
|
data.create_rows("a".into(), 6, 16, "u32".into()),
|
||||||
|
Err("ROWS_PENDING")
|
||||||
|
);
|
||||||
|
assert_eq!(data.delete_rows("a"), Err("ROWS_PENDING"));
|
||||||
|
assert!(!data.free.take(old.offset - data.base, old.bytes));
|
||||||
|
|
||||||
|
// The writer keeps writing the old rows until it copies them over and commits.
|
||||||
|
data.arena[base + 4..base + 8].fill(0xEF);
|
||||||
|
let target = (staged.offset - data.base) as usize + data.arena_start;
|
||||||
|
data.arena.copy_within(base..base + 16, target);
|
||||||
|
data.commit(&["a".into()]).unwrap();
|
||||||
|
assert_eq!(data.rows("a").unwrap().offset, staged.offset);
|
||||||
|
assert_eq!(
|
||||||
|
&data.bytes("a").unwrap()[..8],
|
||||||
|
&[0xCD, 0xCD, 0xCD, 0xCD, 0xEF, 0xEF, 0xEF, 0xEF]
|
||||||
|
);
|
||||||
assert!(data.bytes("a").unwrap()[16..].iter().all(|byte| *byte == 0));
|
assert!(data.bytes("a").unwrap()[16..].iter().all(|byte| *byte == 0));
|
||||||
assert_eq!(data.allocate_object("a", 4).unwrap(), (1, false));
|
assert!(data.free.take(old.offset - data.base, old.bytes));
|
||||||
assert_eq!(data.allocate_object("a", 1).unwrap(), (5, true));
|
assert_eq!(data.commit(&["a".into()]), Err("ROWS_PENDING"));
|
||||||
|
|
||||||
|
// Newly staged rows already feed the span allocator; the writer owns the staged region.
|
||||||
|
data.create_rows("c".into(), 1, 16, "u32".into()).unwrap();
|
||||||
|
assert_eq!(data.allocate_object("a", 4).unwrap(), (1, false, false));
|
||||||
|
let (id, grew, moved) = data.allocate_object("a", 1).unwrap();
|
||||||
|
assert_eq!((id, grew, moved), (5, true, true));
|
||||||
|
assert_eq!(data.staged("a").unwrap().rows, 10);
|
||||||
|
assert_eq!(data.rows("a").unwrap().rows, 5);
|
||||||
|
data.commit(&["a".into()]).unwrap();
|
||||||
|
assert_eq!(rows_of(&data, "a"), 10);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn discarding_a_staged_relocation_restores_capacity() {
|
||||||
|
let mut data = RenderData::new(4096).unwrap();
|
||||||
|
data.create_rows("a".into(), 2, 16, "u32".into()).unwrap();
|
||||||
|
data.create_rows("b".into(), 1, 16, "u32".into()).unwrap();
|
||||||
|
assert_eq!(data.allocate_object("a", 2).unwrap(), (0, false, false));
|
||||||
|
let (staged, moved) = data.create_rows("a".into(), 8, 16, "u32".into()).unwrap();
|
||||||
|
assert!(moved);
|
||||||
|
|
||||||
|
data.discard_pending();
|
||||||
|
assert_eq!(rows_of(&data, "a"), 2);
|
||||||
|
assert!(data.free.take(staged.offset - data.base, staged.bytes));
|
||||||
|
// Rows 2..8 must not be handed out: they only existed in the discarded region.
|
||||||
|
assert_eq!(data.allocate_object("a", 1).unwrap(), (2, true, false));
|
||||||
|
assert_eq!(rows_of(&data, "a"), 4);
|
||||||
}
|
}
|
||||||
|
|||||||
+20
-1
@@ -2,10 +2,13 @@ import initWasm, { Core } from "./pkg/yawn_core.js";
|
|||||||
|
|
||||||
let core;
|
let core;
|
||||||
let profileTimer;
|
let profileTimer;
|
||||||
|
// While a relocation is staged, every request except its commit waits here.
|
||||||
|
let awaitingCommit = false;
|
||||||
|
const queue = [];
|
||||||
|
|
||||||
const fail = code => { throw new Error(code); };
|
const fail = code => { throw new Error(code); };
|
||||||
|
|
||||||
addEventListener("message", async ({ data: message }) => {
|
async function handle(message) {
|
||||||
try {
|
try {
|
||||||
let result;
|
let result;
|
||||||
if (message?.type === "init") {
|
if (message?.type === "init") {
|
||||||
@@ -21,15 +24,22 @@ addEventListener("message", async ({ data: message }) => {
|
|||||||
switch (message?.type) {
|
switch (message?.type) {
|
||||||
case "create-rows":
|
case "create-rows":
|
||||||
result = JSON.parse(core.create_rows(message.name, message.rows, message.stride, message.format));
|
result = JSON.parse(core.create_rows(message.name, message.rows, message.stride, message.format));
|
||||||
|
awaitingCommit = result.commit;
|
||||||
break;
|
break;
|
||||||
case "create-rows-batch":
|
case "create-rows-batch":
|
||||||
result = JSON.parse(core.create_rows_batch(JSON.stringify(message.rows)));
|
result = JSON.parse(core.create_rows_batch(JSON.stringify(message.rows)));
|
||||||
|
awaitingCommit = result.commit.length > 0;
|
||||||
|
break;
|
||||||
|
case "commit-rows":
|
||||||
|
awaitingCommit = false;
|
||||||
|
core.commit_rows(JSON.stringify(message.names));
|
||||||
break;
|
break;
|
||||||
case "delete-rows":
|
case "delete-rows":
|
||||||
core.delete_rows(message.name);
|
core.delete_rows(message.name);
|
||||||
break;
|
break;
|
||||||
case "allocate-object":
|
case "allocate-object":
|
||||||
result = JSON.parse(core.allocate_object(message.name, message.count ?? 1));
|
result = JSON.parse(core.allocate_object(message.name, message.count ?? 1));
|
||||||
|
awaitingCommit = result.commit;
|
||||||
break;
|
break;
|
||||||
case "delete-object":
|
case "delete-object":
|
||||||
core.delete_object(message.name, message.id);
|
core.delete_object(message.name, message.id);
|
||||||
@@ -74,4 +84,13 @@ addEventListener("message", async ({ data: message }) => {
|
|||||||
} catch (error) {
|
} catch (error) {
|
||||||
postMessage({ request: message?.request, error: error?.message ?? "CORE_ERROR" });
|
postMessage({ request: message?.request, error: error?.message ?? "CORE_ERROR" });
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
addEventListener("message", async ({ data: message }) => {
|
||||||
|
if (awaitingCommit && message?.type !== "commit-rows") {
|
||||||
|
queue.push(message);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await handle(message);
|
||||||
|
while (!awaitingCommit && queue.length) await handle(queue.shift());
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -25,7 +25,11 @@ function rowFor(buffer, d, index) {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
The buffer stays the same, but a growing row array may relocate. Replace the cached descriptor—and recreate cached views—after every `create-rows`, batch result, and `allocate-object` result. `allocate-object` always returns `{ id, rows: descriptor }`, even without growth. Never infer an address from a prior descriptor.
|
The buffer stays the same, but a growing row array may relocate. Replace the cached descriptor—and recreate cached views—after every `create-rows`, batch result, and `allocate-object` result. `allocate-object` always returns `{ id, rows: descriptor, commit }`, even without growth. Never infer an address from a prior descriptor.
|
||||||
|
|
||||||
|
### Relocation is owned by the writer
|
||||||
|
|
||||||
|
The worker never copies rows. When growth cannot extend in place it reserves a zeroed region, returns that descriptor with `commit: true` (`create-rows`, `allocate-object`) or the moved names in `commit` (`create-rows-batch`), and keeps rendering from—and the arena owning—the old rows. The main thread is the only writer, so it is the only party that knows its last write landed: on receiving the result it copies the old bytes into the new region, swaps its descriptor, and sends `commit-rows { names }`. The worker then publishes the new descriptor, frees the old rows, and rebuilds the GPU buffers bound to them. Until that commit arrives the worker queues every other request, so a staged relocation is a short, exclusive transaction and no SAB write made at any moment is lost. `index.js` does all of this inside `createRows`, `createRowsBatch`, and `allocateObject`; a raw client must copy and commit synchronously in the same task that received the result.
|
||||||
|
|
||||||
## Slots
|
## Slots
|
||||||
|
|
||||||
|
|||||||
@@ -23,12 +23,13 @@ Worker failures are string codes in `{ request, error }`. GPU/browser validation
|
|||||||
| `ROWS_BUILTIN` | Attempt to grow/delete/allocate from `signals`. |
|
| `ROWS_BUILTIN` | Attempt to grow/delete/allocate from `signals`. |
|
||||||
| `ROWS_UNKNOWN` | Row name does not exist (also possible during frame upload). |
|
| `ROWS_UNKNOWN` | Row name does not exist (also possible during frame upload). |
|
||||||
| `ROWS_ACTIVE` | Delete blocked by active object IDs or active graph reference. |
|
| `ROWS_ACTIVE` | Delete blocked by active object IDs or active graph reference. |
|
||||||
|
| `ROWS_PENDING` | A relocation is staged and uncommitted: growing or deleting that array again, or a `commit-rows` that names a non-staged array or leaves one staged. |
|
||||||
| `ARENA_OOM` | Byte multiplication overflow or no aligned free arena block. |
|
| `ARENA_OOM` | Byte multiplication overflow or no aligned free arena block. |
|
||||||
| `OBJECT_LIMIT` | Slot ID cannot increment past u32. |
|
| `OBJECT_LIMIT` | Slot ID cannot increment past u32. |
|
||||||
| `OBJECT_UNKNOWN` | Delete ID is not active for that row array. |
|
| `OBJECT_UNKNOWN` | Delete ID is not active for that row array. |
|
||||||
| `FPS` | FPS exceeds 1000. |
|
| `FPS` | FPS exceeds 1000. |
|
||||||
|
|
||||||
On row growth, consume the returned descriptor before the next access. Batch creation is not rollback-safe. Do not retry allocation blindly after an uncertain transport outcome: it may allocate a second slot.
|
On row growth, consume the returned descriptor before the next access and commit relocations synchronously. In-place batch creations are not rolled back on a later failure. Do not retry allocation blindly after an uncertain transport outcome: it may allocate a second slot.
|
||||||
|
|
||||||
## Graph parse and planning
|
## Graph parse and planning
|
||||||
|
|
||||||
|
|||||||
@@ -5,10 +5,11 @@ All fields below are top-level beside `type` and `request`. Except for `init` an
|
|||||||
| `type` | Fields | Transfer list | Success `result` | Constraints / effects |
|
| `type` | Fields | Transfer list | Success `result` | Constraints / effects |
|
||||||
|---|---|---|---|---|
|
|---|---|---|---|---|
|
||||||
| `init` | `canvas: OffscreenCanvas`, `arenaBytes: number` | `[canvas]` | `{ buffer: SharedArrayBuffer, rows: Descriptor[], features: string[] }` | Once only; initializes WASM/WebGPU. `features` lists optional device features that were available (`timestamp-query`, `indirect-first-instance`). |
|
| `init` | `canvas: OffscreenCanvas`, `arenaBytes: number` | `[canvas]` | `{ buffer: SharedArrayBuffer, rows: Descriptor[], features: string[] }` | Once only; initializes WASM/WebGPU. `features` lists optional device features that were available (`timestamp-query`, `indirect-first-instance`). |
|
||||||
| `create-rows` | `name`, `rows`, `stride`, `format` | — | descriptor | Nonempty name; rows > 0; stride ≥16 and multiple of 16; format f32/u32/i32. Existing name may only retain format/stride and grow. |
|
| `create-rows` | `name`, `rows`, `stride`, `format` | — | `{ rows: descriptor, commit: boolean }` | Nonempty name; rows > 0; stride ≥16 and multiple of 16; format f32/u32/i32. Existing name may only retain format/stride and grow. `commit: true` means the rows moved: copy the old bytes into the descriptor and send `commit-rows`. |
|
||||||
| `create-rows-batch` | `rows: Array<{name,rows,stride,format}>` | — | descriptor array in input order | Nonempty. Sequential, **not transactional**: earlier creations can survive a later failure. Active GPU resources refresh once after successful batch. |
|
| `create-rows-batch` | `rows: Array<{name,rows,stride,format}>` | — | `{ rows: descriptor[] (input order), commit: string[] }` | Nonempty. Sequential; in-place creations/growths survive a later failure, staged relocations are discarded. `commit` lists the arrays that moved. Active GPU resources refresh once for in-place growth. |
|
||||||
|
| `commit-rows` | `names: string[]` | — | `undefined` | Publishes staged relocations after the caller copied them. Must name every staged array. Frees the old rows, refreshes bound GPU buffers once, sets `sabDirty`. While a relocation is staged the worker queues every other request until this arrives. |
|
||||||
| `delete-rows` | `name` | — | `undefined` | Not `signals`, not active slots, not referenced by active graph. |
|
| `delete-rows` | `name` | — | `undefined` | Not `signals`, not active slots, not referenced by active graph. |
|
||||||
| `allocate-object` | `name`, `count: u32 = 1` | — | `{ id, rows: descriptor }` | Not `signals`; `count` > 0; reserves `count` contiguous rows starting at `id`; may grow/relocate. |
|
| `allocate-object` | `name`, `count: u32 = 1` | — | `{ id, rows: descriptor, commit: boolean }` | Not `signals`; `count` > 0; reserves `count` contiguous rows starting at `id`; may grow. `commit` as in `create-rows`. |
|
||||||
| `delete-object` | `name`, `id: u32` | — | `undefined` | ID must be the first row of an active span; the whole span is zeroed. |
|
| `delete-object` | `name`, `id: u32` | — | `undefined` | ID must be the first row of an active span; the whole span is zeroed. |
|
||||||
| `sync-rows` | `name`, `first: u32 = 0`, `count: u32` | — | `undefined` | Uploads rows `first .. first + count` of the array into every active GPU buffer bound to it (any `sync`) and sets `sabDirty`. `ROW_RANGE` when the span exceeds the array. No-op without an active loadout. |
|
| `sync-rows` | `name`, `first: u32 = 0`, `count: u32` | — | `undefined` | Uploads rows `first .. first + count` of the array into every active GPU buffer bound to it (any `sync`) and sets `sabDirty`. `ROW_RANGE` when the span exceeds the array. No-op without an active loadout. |
|
||||||
| `compile-graph` | `serialized: string` | — | graph ID string | Parses and stores; same ID replaces previously stored graph. Does not activate it. |
|
| `compile-graph` | `serialized: string` | — | graph ID string | Parses and stores; same ID replaces previously stored graph. Does not activate it. |
|
||||||
|
|||||||
Reference in New Issue
Block a user