Skip to content

Commit

Permalink
Merge pull request #219 from singnet/issue-166-docs-and-tests
Browse files Browse the repository at this point in the history 
[das#166] Link Creation Docs and Tests
  • Loading branch information
@eddiebrissow
eddiebrissow authored Jan 29, 2025
2 parents 17ba596 + 3dd2269 commit 07a4e42
Show file tree
Hide file tree
Showing 17 changed files with 581 additions and 302 deletions.
6 changes: 5 additions & 1 deletion src/cpp/link_creation_agent/Makefile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,4 +5,8 @@ build:
cd ../../ && bash scripts/build.sh && cd -

run:
cd ../../ && bash scripts/container_tty.sh ./bin/link_creation_server ${OPTIONS} && cd -
cd ../../ && bash scripts/container_tty.sh ./bin/link_creation_server ${OPTIONS} && cd -


tests:
cd ../../ && bash scripts/unit_tests.sh && cd -
59 changes: 53 additions & 6 deletions src/cpp/link_creation_agent/README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,22 @@

DAS Link Creation Agent (DAS LCA), process a query and create links using the result of the query and a custom template. The service can execute n times a request to update the database with relevant links over the time.

![alt](doc/assets/das_link_creation_hla.png)
### Request

Query Example
The request must have 6 elements:
1. Query (string): A valid Query Agent query (see Query Example)
2. Link Create Template (string): A valid link create template (see Link Create Template Example)
3. Max number of query response (int)
4. Repeat (int): 0 to run once, -1 to run infinitely, 1 or higher to run this number of times.
5. Context (string): Query context for Attention Broker requests
6. Update Attention Broker flag (bool): true or false to update or not the Attention Broker

Example:
```
LINK_TEMPLATE", "Expression", "3", "NODE", "Symbol", "Similarity","VARIABLE", "V1", "VARIABLE", "V2", "LINK_CREATE", "Similarity", "2", "1", "VARIABLE", "V1", "VARIABLE", "V2", "CUSTOM_FIELD", "truth_value", "2", "CUSTOM_FIELD", "mean", "2", "count", "10", "avg", "0.9", "confidence", "0.9", "10", "0", "test", "false"
```

#### Query Example

```
LINK_TEMPLATE Expression 3
Expand All @@ -20,7 +33,7 @@ LINK_TEMPLATE Expression 3
VARIABLE V2
```

Example of template for link creation:
#### Link Create Template Example:

```
LINK_CREATE Similarity 2 1
Expand All @@ -32,13 +45,47 @@ LINK_CREATE Similarity 2 1
avg 0.9
confidence 0.9
```
#### CUSTOM FIELDS
TBD
A link creation request is expected to have 4 elements:
1. LINK_CREATE: required keyword
2. Link type: the type of links that this request will create
3. Link targets count: the count of link targets
4. Custom field count: the count of custom fields

CUSTOM FIELDS

A custom field must have 3 elements:
1. CUSTOM_FIELD: required keyword
2. Custom field name
3. Custom field values count: the count of the values


## How to build

```
make build
```

## Development
## How to run
Running the server
```
make run OPTIONS="--type server --config_file <path_to_config_file>"
```

#### Config file example
```
default_interval = 10
thread_count = 5
query_node_server_id = localhost:35700
query_node_client_id = localhost:9001
link_creation_server_id = localhost:9080
das_client_id = localhost:9090
requests_buffer_file = ./buffer
```
* default_interval: The default interval to run requests that repeat one or more times
* thread_count: The number of threads to process Query Agent requests
* query_node_server_id: IP + port of the Query Agent server.
* query_node_client_id: IP + port of the Query Agent client (local machine)
* link_creation_server_id: IP + port of the Link Creation Agent
* das_client_id: IP + port of DAS to add links
* requests_buffer_file: path to request buffer, where are stored all requests that repeat

97 changes: 51 additions & 46 deletions src/cpp/link_creation_agent/agent.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,10 @@ using namespace query_element;
LinkCreationAgent::LinkCreationAgent(string config_path) {
this->config_path = config_path;
load_config();
link_creation_node_server = new LinkCreationNode(link_creation_server_id);
query_node_client = new DASNode(query_node_client_id, query_node_server_id);
service = new LinkCreationService(thread_count);
das_client = new ServerNode(das_client_id, query_node_server_id);
link_creation_node_server = new LinkCreationNode(link_creation_agent_server_id);
query_node_client = new DASNode(query_agent_client_id, query_agent_server_id);
service = new LinkCreationService(link_creation_agent_thread_count);
das_client = new DasAgentNode(das_agent_client_id, das_agent_server_id);

this->agent_thread = new thread(&LinkCreationAgent::run, this);
}
Expand All @@ -30,7 +30,12 @@ LinkCreationAgent::~LinkCreationAgent() {
}

void LinkCreationAgent::stop() {
save_buffer();
agent_mutex.lock();
if(!is_stoping){
is_stoping = true;
save_buffer();
}
agent_mutex.unlock();
link_creation_node_server->graceful_shutdown();
query_node_client->graceful_shutdown();
das_client->graceful_shutdown();
Expand All @@ -43,57 +48,56 @@ void LinkCreationAgent::stop() {
void LinkCreationAgent::run() {
int current_buffer_position = 0;
while (true) {
if(is_stoping) break;
LinkCreationAgentRequest* lca_request = NULL;
bool is_from_buffer = false;
if (!link_creation_node_server->is_query_empty()) {
vector<string> request = link_creation_node_server->pop_request();
lca_request = handle_request(request);
lca_request = create_request(request);
lca_request->current_interval = requests_interval_seconds;
if (lca_request != NULL && (lca_request->infinite || lca_request->repeat > 0)) {
request_buffer.push_back(*lca_request);
}
} else {
if (!request_buffer.empty()) {
is_from_buffer = true;
current_buffer_position = current_buffer_position % request_buffer.size();
lca_request = &request_buffer[current_buffer_position];
current_buffer_position++;
}
}

if (lca_request == NULL) {
this_thread::sleep_for(chrono::seconds(loop_interval));
if (lca_request == NULL ||
lca_request->last_execution + lca_request->current_interval > time(0)) {
this_thread::sleep_for(chrono::milliseconds(loop_interval));
continue;
}

if (lca_request->last_execution == 0 ||
lca_request->last_execution + lca_request->current_interval < time(0)) {
shared_ptr<RemoteIterator> iterator =
query(lca_request->query, lca_request->context, lca_request->update_attention_broker);
// cout << "Request query: " << lca_request->query[0] << endl;
// cout << "Request link_template: " << lca_request->link_template[0] << endl;
// cout << "Request max_results: " << lca_request->max_results << endl;
// cout << "Request repeat: " << lca_request->repeat << endl;
// cout << "Request current_interval: " << lca_request->current_interval << endl;
// cout << "Request last_execution: " << lca_request->last_execution << endl;

service->process_request(iterator, das_client, lca_request->link_template);

if (lca_request->infinite || lca_request->repeat > 0) {
lca_request->last_execution = time(0);
lca_request->current_interval = (lca_request->current_interval * 2) % 86400; // Add exponential backoff, resets after 24 hours
if (lca_request->repeat > 1) {
lca_request->repeat--;
} else {
// TODO check for memory leaks here
request_buffer.erase(request_buffer.begin() + current_buffer_position - 1);
}
shared_ptr<RemoteIterator> iterator =
query(lca_request->query, lca_request->context, lca_request->update_attention_broker);

service->process_request(
iterator, das_client, lca_request->link_template, lca_request->max_results);

if (lca_request->infinite || lca_request->repeat > 0) {
lca_request->last_execution = time(0);
lca_request->current_interval =
(lca_request->current_interval * 2) %
86400; // TODO Add exponential backoff, resets after 24 hours

if (lca_request->infinite) continue;
if (lca_request->repeat >= 1) lca_request->repeat--;

} else {
if (is_from_buffer) {
request_buffer.erase(request_buffer.begin() + current_buffer_position - 1);
} else {
delete lca_request;
}
}
}
}

void LinkCreationAgent::clean_requests() {}

shared_ptr<RemoteIterator> LinkCreationAgent::query(vector<string>& query_tokens,
string context,
bool update_attention_broker) {
Expand All @@ -112,20 +116,22 @@ void LinkCreationAgent::load_config() {
if (getline(is_line, value)) {
value.erase(remove(value.begin(), value.end(), ' '), value.end());
key.erase(remove(key.begin(), key.end(), ' '), key.end());
if (key == "default_interval")
this->default_interval = stoi(value);
else if (key == "thread_count")
this->thread_count = stoi(value);
else if (key == "query_node_client_id")
this->query_node_client_id = value;
else if (key == "query_node_server_id")
this->query_node_server_id = value;
else if (key == "link_creation_server_id")
this->link_creation_server_id = value;
else if (key == "das_client_id")
this->das_client_id = value;
if (key == "requests_interval_seconds")
this->requests_interval_seconds = stoi(value);
else if (key == "link_creation_agent_thread_count")
this->link_creation_agent_thread_count = stoi(value);
else if (key == "query_agent_client_id")
this->query_agent_client_id = value;
else if (key == "query_agent_server_id")
this->query_agent_server_id = value;
else if (key == "link_creation_agent_server_id")
this->link_creation_agent_server_id = value;
else if (key == "das_agent_client_id")
this->das_agent_client_id = value;
else if (key == "requests_buffer_file")
this->requests_buffer_file = value;
else if (key == "das_agent_server_id")
this->das_agent_server_id = value;
else if (key == "context")
this->context = value;
}
Expand All @@ -152,7 +158,7 @@ void LinkCreationAgent::load_buffer() {
file.close();
}

LinkCreationAgentRequest* LinkCreationAgent::handle_request(vector<string> request) {
LinkCreationAgentRequest* LinkCreationAgent::create_request(vector<string> request) {
try {
LinkCreationAgentRequest* lca_request = new LinkCreationAgentRequest();
int cursor = 0;
Expand Down Expand Up @@ -180,7 +186,6 @@ LinkCreationAgentRequest* LinkCreationAgent::handle_request(vector<string> reque
}
}
lca_request->infinite = (lca_request->repeat == -1);
lca_request->current_interval = default_interval;

return lca_request;
} catch (exception& e) {
Expand Down
Loading

0 comments on commit 07a4e42

Please sign in to comment.