more release lifetime cleanup and base class contextmanager stuff
This commit is contained in:
parent
1f777a73b5
commit
2314fcb57e
5
.vscode/settings.json
vendored
5
.vscode/settings.json
vendored
@ -1,3 +1,6 @@
|
|||||||
{
|
{
|
||||||
"cmake.configureOnOpen": false
|
"cmake.configureOnOpen": false,
|
||||||
|
"python.analysis.extraPaths": [
|
||||||
|
"veilid-python/.venv/lib/python3.11/site-packages"
|
||||||
|
],
|
||||||
}
|
}
|
@ -233,8 +233,11 @@ impl StorageManager {
|
|||||||
force_refresh: bool,
|
force_refresh: bool,
|
||||||
) -> VeilidAPIResult<Option<ValueData>> {
|
) -> VeilidAPIResult<Option<ValueData>> {
|
||||||
let mut inner = self.lock().await?;
|
let mut inner = self.lock().await?;
|
||||||
let Some(opened_record) = inner.opened_records.remove(&key) else {
|
let safety_selection = {
|
||||||
apibail_generic!("record not open");
|
let Some(opened_record) = inner.opened_records.get(&key) else {
|
||||||
|
apibail_generic!("record not open");
|
||||||
|
};
|
||||||
|
opened_record.safety_selection()
|
||||||
};
|
};
|
||||||
|
|
||||||
// See if the requested subkey is our local record store
|
// See if the requested subkey is our local record store
|
||||||
@ -269,7 +272,7 @@ impl StorageManager {
|
|||||||
rpc_processor,
|
rpc_processor,
|
||||||
key,
|
key,
|
||||||
subkey,
|
subkey,
|
||||||
opened_record.safety_selection(),
|
safety_selection,
|
||||||
last_subkey_result,
|
last_subkey_result,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
@ -307,12 +310,18 @@ impl StorageManager {
|
|||||||
apibail_generic!("unsupported cryptosystem");
|
apibail_generic!("unsupported cryptosystem");
|
||||||
};
|
};
|
||||||
|
|
||||||
let Some(opened_record) = inner.opened_records.remove(&key) else {
|
let (safety_selection, opt_writer) = {
|
||||||
apibail_generic!("record not open");
|
let Some(opened_record) = inner.opened_records.get(&key) else {
|
||||||
|
apibail_generic!("record not open");
|
||||||
|
};
|
||||||
|
(
|
||||||
|
opened_record.safety_selection(),
|
||||||
|
opened_record.writer().cloned(),
|
||||||
|
)
|
||||||
};
|
};
|
||||||
|
|
||||||
// If we don't have a writer then we can't write
|
// If we don't have a writer then we can't write
|
||||||
let Some(writer) = opened_record.writer().cloned() else {
|
let Some(writer) = opt_writer else {
|
||||||
apibail_generic!("value is not writable");
|
apibail_generic!("value is not writable");
|
||||||
};
|
};
|
||||||
|
|
||||||
@ -371,7 +380,7 @@ impl StorageManager {
|
|||||||
rpc_processor,
|
rpc_processor,
|
||||||
key,
|
key,
|
||||||
subkey,
|
subkey,
|
||||||
opened_record.safety_selection(),
|
safety_selection,
|
||||||
signed_value_data,
|
signed_value_data,
|
||||||
descriptor,
|
descriptor,
|
||||||
)
|
)
|
||||||
|
@ -428,7 +428,7 @@ impl TableStore {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn on_table_db_drop(&self, table: String) {
|
pub(crate) fn on_table_db_drop(&self, table: String) {
|
||||||
log_rtab!(debug "dropping table db: {}", table);
|
log_rtab!("dropping table db: {}", table);
|
||||||
let mut inner = self.inner.lock();
|
let mut inner = self.inner.lock();
|
||||||
if inner.opened.remove(&table).is_none() {
|
if inner.opened.remove(&table).is_none() {
|
||||||
unreachable!("should have removed an item");
|
unreachable!("should have removed an item");
|
||||||
|
@ -1,13 +1,13 @@
|
|||||||
# Routing context veilid tests
|
# # Routing context veilid tests
|
||||||
|
|
||||||
import veilid
|
# import veilid
|
||||||
import pytest
|
# import pytest
|
||||||
import asyncio
|
# import asyncio
|
||||||
import json
|
# import json
|
||||||
from . import *
|
# from . import *
|
||||||
|
|
||||||
##################################################################
|
# ##################################################################
|
||||||
BOGUS_KEY = veilid.TypedKey.from_value(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.PublicKey.from_bytes(b' '))
|
# BOGUS_KEY = veilid.TypedKey.from_value(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.PublicKey.from_bytes(b' '))
|
||||||
|
|
||||||
# @pytest.mark.asyncio
|
# @pytest.mark.asyncio
|
||||||
# async def test_get_dht_value_unopened(api_connection: veilid.VeilidAPI):
|
# async def test_get_dht_value_unopened(api_connection: veilid.VeilidAPI):
|
||||||
@ -46,6 +46,29 @@ BOGUS_KEY = veilid.TypedKey.from_value(veilid.CryptoKind.CRYPTO_KIND_VLD0, veili
|
|||||||
# await rc.close_dht_record(rec.key)
|
# await rc.close_dht_record(rec.key)
|
||||||
# await rc.delete_dht_record(rec.key)
|
# await rc.delete_dht_record(rec.key)
|
||||||
|
|
||||||
# xxx make tests for tabledb api first
|
# @pytest.mark.asyncio
|
||||||
# xxx then make a test that creates a record, stores it in a table
|
# async def test_get_dht_value_nonexistent(api_connection: veilid.VeilidAPI):
|
||||||
# xxx then make another test that gets the keys from the table and closes/deletes them
|
# rc = await api_connection.new_routing_context()
|
||||||
|
# async with rc:
|
||||||
|
# rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
|
||||||
|
# assert await rc.get_dht_value(rec.key, 0, False) == None
|
||||||
|
# await rc.close_dht_record(rec.key)
|
||||||
|
# await rc.delete_dht_record(rec.key)
|
||||||
|
|
||||||
|
# @pytest.mark.asyncio
|
||||||
|
# async def test_set_get_dht_value(api_connection: veilid.VeilidAPI):
|
||||||
|
# rc = await api_connection.new_routing_context()
|
||||||
|
# async with rc:
|
||||||
|
# rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
|
||||||
|
|
||||||
|
# vd = await rc.set_dht_value(rec.key, 0, b"BLAH BLAH BLAH")
|
||||||
|
# assert vd != None
|
||||||
|
|
||||||
|
# vd2 = await rc.get_dht_value(rec.key, 0, False)
|
||||||
|
# assert vd2 != None
|
||||||
|
|
||||||
|
# assert vd == vd2
|
||||||
|
|
||||||
|
# await rc.close_dht_record(rec.key)
|
||||||
|
# await rc.delete_dht_record(rec.key)
|
||||||
|
|
||||||
|
@ -6,6 +6,18 @@ from .state import VeilidState
|
|||||||
|
|
||||||
|
|
||||||
class RoutingContext(ABC):
|
class RoutingContext(ABC):
|
||||||
|
|
||||||
|
async def __aenter__(self) -> Self:
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *excinfo):
|
||||||
|
if not self.is_done():
|
||||||
|
await self.release()
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def is_done(self) -> bool:
|
||||||
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
async def release(self):
|
async def release(self):
|
||||||
pass
|
pass
|
||||||
@ -84,6 +96,17 @@ class RoutingContext(ABC):
|
|||||||
|
|
||||||
|
|
||||||
class TableDbTransaction(ABC):
|
class TableDbTransaction(ABC):
|
||||||
|
async def __aenter__(self) -> Self:
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *excinfo):
|
||||||
|
if not self.is_done():
|
||||||
|
await self.rollback()
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def is_done(self) -> bool:
|
||||||
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
async def commit(self):
|
async def commit(self):
|
||||||
pass
|
pass
|
||||||
@ -102,6 +125,17 @@ class TableDbTransaction(ABC):
|
|||||||
|
|
||||||
|
|
||||||
class TableDb(ABC):
|
class TableDb(ABC):
|
||||||
|
async def __aenter__(self) -> Self:
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *excinfo):
|
||||||
|
if not self.is_done():
|
||||||
|
await self.release()
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def is_done(self) -> bool:
|
||||||
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
async def release(self):
|
async def release(self):
|
||||||
pass
|
pass
|
||||||
@ -132,6 +166,18 @@ class TableDb(ABC):
|
|||||||
|
|
||||||
|
|
||||||
class CryptoSystem(ABC):
|
class CryptoSystem(ABC):
|
||||||
|
|
||||||
|
async def __aenter__(self) -> Self:
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *excinfo):
|
||||||
|
if not self.is_done():
|
||||||
|
await self.release()
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def is_done(self) -> bool:
|
||||||
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
async def release(self):
|
async def release(self):
|
||||||
pass
|
pass
|
||||||
@ -246,6 +292,21 @@ class CryptoSystem(ABC):
|
|||||||
|
|
||||||
|
|
||||||
class VeilidAPI(ABC):
|
class VeilidAPI(ABC):
|
||||||
|
async def __aenter__(self) -> Self:
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aexit__(self, *excinfo):
|
||||||
|
if not self.is_done():
|
||||||
|
await self.release()
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def is_done(self) -> bool:
|
||||||
|
pass
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
async def release(self):
|
||||||
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
async def control(self, args: list[str]) -> str:
|
async def control(self, args: list[str]) -> str:
|
||||||
pass
|
pass
|
||||||
|
@ -53,7 +53,8 @@ class _JsonVeilidAPI(VeilidAPI):
|
|||||||
writer: Optional[asyncio.StreamWriter]
|
writer: Optional[asyncio.StreamWriter]
|
||||||
update_callback: Callable[[VeilidUpdate], Awaitable]
|
update_callback: Callable[[VeilidUpdate], Awaitable]
|
||||||
handle_recv_messages_task: Optional[asyncio.Task]
|
handle_recv_messages_task: Optional[asyncio.Task]
|
||||||
validate_schemas: bool
|
validate_schema: bool
|
||||||
|
done: bool
|
||||||
# Shared Mutable State
|
# Shared Mutable State
|
||||||
lock: asyncio.Lock
|
lock: asyncio.Lock
|
||||||
next_id: int
|
next_id: int
|
||||||
@ -70,17 +71,12 @@ class _JsonVeilidAPI(VeilidAPI):
|
|||||||
self.writer = writer
|
self.writer = writer
|
||||||
self.update_callback = update_callback
|
self.update_callback = update_callback
|
||||||
self.validate_schema = validate_schema
|
self.validate_schema = validate_schema
|
||||||
|
self.done = False
|
||||||
self.handle_recv_messages_task = None
|
self.handle_recv_messages_task = None
|
||||||
self.lock = asyncio.Lock()
|
self.lock = asyncio.Lock()
|
||||||
self.next_id = 1
|
self.next_id = 1
|
||||||
self.in_flight_requests = dict()
|
self.in_flight_requests = dict()
|
||||||
|
|
||||||
async def __aenter__(self) -> Self:
|
|
||||||
return self
|
|
||||||
|
|
||||||
async def __aexit__(self, *excinfo):
|
|
||||||
await self.close()
|
|
||||||
|
|
||||||
async def _cleanup_close(self):
|
async def _cleanup_close(self):
|
||||||
await self.lock.acquire()
|
await self.lock.acquire()
|
||||||
try:
|
try:
|
||||||
@ -96,7 +92,10 @@ class _JsonVeilidAPI(VeilidAPI):
|
|||||||
finally:
|
finally:
|
||||||
self.lock.release()
|
self.lock.release()
|
||||||
|
|
||||||
async def close(self):
|
def is_done(self) -> bool:
|
||||||
|
return self.done
|
||||||
|
|
||||||
|
async def release(self):
|
||||||
# Take the task
|
# Take the task
|
||||||
await self.lock.acquire()
|
await self.lock.acquire()
|
||||||
try:
|
try:
|
||||||
@ -112,6 +111,7 @@ class _JsonVeilidAPI(VeilidAPI):
|
|||||||
await handle_recv_messages_task
|
await handle_recv_messages_task
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
pass
|
pass
|
||||||
|
self.done = True
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def connect(
|
async def connect(
|
||||||
@ -430,12 +430,8 @@ class _JsonRoutingContext(RoutingContext):
|
|||||||
# complain
|
# complain
|
||||||
raise AssertionError("Should have released routing context before dropping object")
|
raise AssertionError("Should have released routing context before dropping object")
|
||||||
|
|
||||||
async def __aenter__(self) -> Self:
|
def is_done(self) -> bool:
|
||||||
return self
|
return self.done
|
||||||
|
|
||||||
async def __aexit__(self, *excinfo):
|
|
||||||
if not self.done:
|
|
||||||
await self.release()
|
|
||||||
|
|
||||||
async def release(self):
|
async def release(self):
|
||||||
if self.done:
|
if self.done:
|
||||||
@ -668,12 +664,8 @@ class _JsonTableDbTransaction(TableDbTransaction):
|
|||||||
# complain
|
# complain
|
||||||
raise AssertionError("Should have committed or rolled back transaction before dropping object")
|
raise AssertionError("Should have committed or rolled back transaction before dropping object")
|
||||||
|
|
||||||
async def __aenter__(self) -> Self:
|
def is_done(self) -> bool:
|
||||||
return self
|
return self.done
|
||||||
|
|
||||||
async def __aexit__(self, *excinfo):
|
|
||||||
if not self.done:
|
|
||||||
await self.rollback()
|
|
||||||
|
|
||||||
async def commit(self):
|
async def commit(self):
|
||||||
if self.done:
|
if self.done:
|
||||||
@ -753,12 +745,8 @@ class _JsonTableDb(TableDb):
|
|||||||
# complain
|
# complain
|
||||||
raise AssertionError("Should have released table db before dropping object")
|
raise AssertionError("Should have released table db before dropping object")
|
||||||
|
|
||||||
async def __aenter__(self) -> Self:
|
def is_done(self) -> bool:
|
||||||
return self
|
return self.done
|
||||||
|
|
||||||
async def __aexit__(self, *excinfo):
|
|
||||||
if not self.done:
|
|
||||||
await self.release()
|
|
||||||
|
|
||||||
async def release(self):
|
async def release(self):
|
||||||
if self.done:
|
if self.done:
|
||||||
@ -880,13 +868,9 @@ class _JsonCryptoSystem(CryptoSystem):
|
|||||||
# complain
|
# complain
|
||||||
raise AssertionError("Should have released crypto system before dropping object")
|
raise AssertionError("Should have released crypto system before dropping object")
|
||||||
|
|
||||||
async def __aenter__(self) -> Self:
|
def is_done(self) -> bool:
|
||||||
return self
|
return self.done
|
||||||
|
|
||||||
async def __aexit__(self, *excinfo):
|
|
||||||
if not self.done:
|
|
||||||
await self.release()
|
|
||||||
|
|
||||||
async def release(self):
|
async def release(self):
|
||||||
if self.done:
|
if self.done:
|
||||||
return
|
return
|
||||||
|
Loading…
Reference in New Issue
Block a user