Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[MEMORY]: Improve memory syncing #1449

Merged
merged 2 commits into from
Nov 26, 2024
Merged

[MEMORY]: Improve memory syncing #1449

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
859e2d8
[MEMORY]: Improve memory syncing
amadolid Nov 18, 2024
36842fb
[UNIT-TEST]: Memory sync issue sample test
amadolid Nov 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 31 additions & 17 deletions jac-cloud/jac_cloud/core/memory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,23 +114,14 @@ def close(self) -> None:

super().close()

def get_bulk_write(self) -> BulkWrite:
"""Sync memory to database."""
bulk_write = BulkWrite()

for anchor in self.__gc__:
match anchor:
case NodeAnchor():
bulk_write.del_node(anchor.id)
case EdgeAnchor():
bulk_write.del_edge(anchor.id)
case WalkerAnchor():
bulk_write.del_walker(anchor.id)
case _:
pass

for anchor in self.__mem__.values():
if anchor.architype and anchor.persistent:
def sync_mem_to_db(self, bulk_write: BulkWrite, keys: Iterable[ObjectId]) -> None:
"""Manually sync memory to db."""
for key in keys:
if (
(anchor := self.__mem__.get(key))
and anchor.architype
and anchor.persistent
):
if not anchor.state.connected:
anchor.state.connected = True
anchor.sync_hash()
Expand All @@ -151,4 +142,27 @@ def get_bulk_write(self) -> BulkWrite:
else:
anchor.update(bulk_write)

def get_bulk_write(self) -> BulkWrite:
"""Sync memory to database."""
bulk_write = BulkWrite()

for anchor in self.__gc__:
match anchor:
case NodeAnchor():
bulk_write.del_node(anchor.id)
case EdgeAnchor():
bulk_write.del_edge(anchor.id)
case WalkerAnchor():
bulk_write.del_walker(anchor.id)
case _:
pass

keys = set(self.__mem__.keys())

# current memory
self.sync_mem_to_db(bulk_write, keys)

# additional after memory sync
self.sync_mem_to_db(bulk_write, set(self.__mem__.keys() - keys))

return bulk_write
80 changes: 80 additions & 0 deletions jac-cloud/jac_cloud/tests/openapi_specs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -382,6 +382,15 @@ components:
- root_id
title: allow_other_root_access_body_model
type: object
check_memory_sync_body_model:
properties:
other_node_id:
title: Other Node Id
type: string
required:
- other_node_id
title: check_memory_sync_body_model
type: object
combination1_body_model:
properties:
c:
Expand Down Expand Up @@ -841,6 +850,77 @@ paths:
tags:
- walker
- walker
/walker/check_memory_sync:
post:
operationId: api_root_walker_check_memory_sync_post
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/check_memory_sync_body_model'
required: true
responses:
'200':
content:
application/json:
schema:
anyOf:
- $ref: '#/components/schemas/ContextResponse_NoneType_'
- {}
title: Response Api Root Walker Check Memory Sync Post
description: Successful Response
'422':
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
description: Validation Error
security:
- HTTPBearer: []
summary: /check_memory_sync
tags:
- walker
- walker
/walker/check_memory_sync/{node}:
post:
operationId: api_entry_walker_check_memory_sync__node__post
parameters:
- in: path
name: node
required: true
schema:
anyOf:
- type: string
- type: 'null'
title: Node
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/check_memory_sync_body_model'
required: true
responses:
'200':
content:
application/json:
schema:
anyOf:
- $ref: '#/components/schemas/ContextResponse_NoneType_'
- {}
title: Response Api Entry Walker Check Memory Sync Node Post
description: Successful Response
'422':
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
description: Validation Error
security:
- HTTPBearer: []
summary: /check_memory_sync/{node}
tags:
- walker
- walker
/walker/check_populated_graph:
post:
operationId: api_root_walker_check_populated_graph_post
Expand Down
10 changes: 10 additions & 0 deletions jac-cloud/jac_cloud/tests/simple_graph.jac
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -710,4 +710,14 @@ walker check_populated_graph {

report count;
}
}

walker check_memory_sync {
has other_node_id: str;
can enter with `root entry {
import:py from jac_cloud.core.architype {NodeAnchor}
n = &(self.other_node_id);
n.val = 4;
report n;
}
}
28 changes: 25 additions & 3 deletions jac-cloud/jac_cloud/tests/test_simple_graph.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -560,19 +560,19 @@ def trigger_upload_file(self) -> None:
"single": {
"name": "simple_graph.jac",
"content_type": "application/octet-stream",
"size": 15146,
"size": 15376,
}
},
"multiple": [
{
"name": "simple_graph.jac",
"content_type": "application/octet-stream",
"size": 15146,
"size": 15376,
},
{
"name": "simple_graph.jac",
"content_type": "application/octet-stream",
"size": 15146,
"size": 15376,
},
],
"singleOptional": None,
Expand Down Expand Up @@ -620,6 +620,22 @@ def trigger_reset_graph(self) -> None:
self.assertEqual([None], res["returns"])
self.assertEqual([1], res["reports"])

def trigger_memory_sync(self) -> None:
"""Test memory sync."""
res = self.post_api("traverse_graph")

self.assertEqual(200, res["status"])
self.assertEqual([None, None, None], res["returns"])

a_node = res["reports"].pop(1)
self.assertTrue(a_node["id"].startswith("n:A:"))
self.assertEqual({"val": 1}, a_node["context"])

res = self.post_api(
"check_memory_sync", json={"other_node_id": a_node["id"]}, user=1
)
self.assertEqual(200, res["status"])

def test_all_features(self) -> None:
"""Test Full Features."""
self.trigger_openapi_specs_test()
Expand Down Expand Up @@ -716,3 +732,9 @@ def test_all_features(self) -> None:
###################################################

self.trigger_reset_graph()

###################################################
# TEST MEMORY SYNC #
###################################################

self.trigger_memory_sync()
Loading