Fix table RPC to not be interruptible
This commit is contained in:
parent
2bea76ce16
commit
43ce5e4ab4
2 changed files with 25 additions and 34 deletions
33
TODO
33
TODO
|
@ -1,29 +1,16 @@
|
||||||
Object table
|
Replication
|
||||||
------------
|
|
||||||
|
|
||||||
|
|
||||||
Rename version table to object table
|
|
||||||
In value handle the different versions
|
|
||||||
|
|
||||||
So that the table becomes bucket + Sort key = object key -> CRDT(list of versions)
|
|
||||||
|
|
||||||
CRDT merge rule:
|
|
||||||
- keep one complete version (the one with the highest timestamp)
|
|
||||||
- keep all incomplete versions with timestamps higher than the complete version
|
|
||||||
|
|
||||||
Cleanup rule: remove incomplete versions after a given delay (say 24h)
|
|
||||||
|
|
||||||
|
|
||||||
Block table
|
|
||||||
-----------
|
-----------
|
||||||
|
|
||||||
Table is version_UUID -> BTreeMap<(offset, block hash)> OR Deleted (= CRDT top)
|
- for each interval of tokens, we know the list of nodes that are responsible
|
||||||
|
- every node watches the current ring and state of the network
|
||||||
|
- and thus determines the interval of tokens for which they are responsible
|
||||||
|
|
||||||
|
|
||||||
Block reference table
|
|
||||||
---------------------
|
|
||||||
|
|
||||||
Table is block_Hash + Sort key: version_UUID -> boolean (true when deleted)
|
To do list
|
||||||
|
----------
|
||||||
|
|
||||||
Since the hash key is the same as for the blocks themselves,
|
- important: check block values on read and repare corrupted block contents
|
||||||
we can simply consider the updates to this table as events that increase/decrease a reference counter.
|
- less a priority: hinted handoff
|
||||||
|
- FIXME in rpc_server when garage shuts down and futures can be interrupted
|
||||||
|
(tokio::spawn should be replaced by a new function background::spawn_joinable)
|
||||||
|
|
|
@ -76,22 +76,26 @@ async fn handler(
|
||||||
// and the request handler simply sits there waiting for the task to finish.
|
// and the request handler simply sits there waiting for the task to finish.
|
||||||
// (if it's cancelled, that's not an issue)
|
// (if it's cancelled, that's not an issue)
|
||||||
// (TODO FIXME except if garage happens to shut down at that point)
|
// (TODO FIXME except if garage happens to shut down at that point)
|
||||||
let write_fut = async move { garage.block_manager.write_block(&m.hash, &m.data).await };
|
let write_fut = async move {
|
||||||
|
garage.block_manager.write_block(&m.hash, &m.data).await
|
||||||
|
};
|
||||||
tokio::spawn(write_fut).await?
|
tokio::spawn(write_fut).await?
|
||||||
}
|
}
|
||||||
Message::GetBlock(h) => garage.block_manager.read_block(&h).await,
|
Message::GetBlock(h) => garage.block_manager.read_block(&h).await,
|
||||||
|
|
||||||
Message::TableRPC(table, msg) => {
|
Message::TableRPC(table, msg) => {
|
||||||
// For now, table RPCs use transactions that are not async so even if the future
|
// Same trick for table RPCs than for PutBlock
|
||||||
// is canceled, the db should be in a consistent state.
|
let op_fut = async move {
|
||||||
if let Some(rpc_handler) = garage.table_rpc_handlers.get(&table) {
|
if let Some(rpc_handler) = garage.table_rpc_handlers.get(&table) {
|
||||||
rpc_handler
|
rpc_handler
|
||||||
.handle(&msg[..])
|
.handle(&msg[..])
|
||||||
.await
|
.await
|
||||||
.map(|rep| Message::TableRPC(table.to_string(), rep))
|
.map(|rep| Message::TableRPC(table.to_string(), rep))
|
||||||
} else {
|
} else {
|
||||||
Ok(Message::Error(format!("Unknown table: {}", table)))
|
Ok(Message::Error(format!("Unknown table: {}", table)))
|
||||||
}
|
}
|
||||||
|
};
|
||||||
|
tokio::spawn(op_fut).await?
|
||||||
}
|
}
|
||||||
|
|
||||||
_ => Ok(Message::Error(format!("Unexpected message: {:?}", msg))),
|
_ => Ok(Message::Error(format!("Unexpected message: {:?}", msg))),
|
||||||
|
|
Loading…
Reference in a new issue