gfwleak/stellar-stellar
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🦄 refactor(plug_mgr): per stage msg counter encap

Browse Source
This commit is contained in:
yangwei
2024-09-06 14:01:23 +08:00
parent 9cb50f81fb
commit 3de8bbdabc
3 changed files with 26 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
infra/plugin_manager/plugin_manager.c
View File

@@ -10,6 +10,24 @@
#include "packet_private.h"
#include "session_private.h"
#include <stdbool.h>
void stellar_per_stage_message_counter_incby(struct plugin_manager_schema *plug_mgr, int tid, long long increment)
{
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt+=increment;
}
void stellar_per_stage_message_counter_set(struct plugin_manager_schema *plug_mgr, int tid, long long increment)
{
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt=increment;
}
bool stellar_per_stage_message_counter_overlimt(struct plugin_manager_schema *plug_mgr, int tid)
{
if(plug_mgr->per_thread_data[tid].pub_packet_msg_cnt >= plug_mgr->max_message_dispatch)return true;
return false;
}
UT_icd plugin_specs_icd = {sizeof(struct plugin_specific), NULL, NULL, NULL};
static struct plugin_specific *plugin_specs_load(const char *toml_conf_path, int *spec_num)
@@ -548,8 +566,7 @@ int stellar_mq_publish_message_with_priority(struct stellar *st, int topic_id, v
if(plug_mgr==NULL || plug_mgr->stellar_mq_schema_array == NULL)return -1;
int tid = stellar_get_current_thread_index();
if(plug_mgr->per_thread_data[tid].pub_packet_msg_cnt == -1)return -1;
if(plug_mgr->per_thread_data[tid].pub_packet_msg_cnt >= plug_mgr->max_message_dispatch)return -1;
if(stellar_per_stage_message_counter_overlimt(plug_mgr, tid)==true)return -1;
unsigned int len = utarray_len(plug_mgr->stellar_mq_schema_array);
if (len <= (unsigned int)topic_id)return -1;
@@ -559,7 +576,7 @@ int stellar_mq_publish_message_with_priority(struct stellar *st, int topic_id, v
msg->header.priority = priority;
msg->body = data;
DL_APPEND(plug_mgr->per_thread_data[tid].priority_mq[priority], msg);
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt+=1;
stellar_per_stage_message_counter_incby(plug_mgr, tid, 1);
return 0;
}
@@ -634,8 +651,7 @@ static void plugin_manager_on_packet(struct plugin_manager_schema *plug_mgr, str
struct registered_plugin_schema *p=NULL;
int tid=stellar_get_current_thread_index();
//TODO : provide public api to reset pub_msg_cnt
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt=0;//reset pub_msg_cnt
stellar_per_stage_message_counter_set(plug_mgr, tid, 0);
while ((p = (struct registered_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, p)))
{
if(p->on_packet[in_out])
@@ -657,7 +673,7 @@ void plugin_manager_on_packet_output(struct plugin_manager_schema *plug_mgr, str
if(plug_mgr == NULL || plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
plugin_manager_on_packet(plug_mgr, pkt, PACKET_STAGE_OUTPUT);
int tid=stellar_get_current_thread_index();
plug_mgr->per_thread_data[tid].pub_packet_msg_cnt=-1;//disable packet message publish
stellar_per_stage_message_counter_set(plug_mgr, tid, -1);
stellar_mq_free(&plug_mgr->per_thread_data[tid].dealth_letter_queue,
plug_mgr->stellar_mq_schema_array);
per_thread_packet_exdata_arrary_clean(plug_mgr);