对libcomm_memory.cpp文件中的函数进行了注释。
This commit is contained in:
parent
45bbe9e64e
commit
b3f0eeb4ef
|
|
@ -65,7 +65,6 @@
|
|||
#define static
|
||||
#endif
|
||||
|
||||
|
||||
#define STREAM_SCAN_FINISH 'F'
|
||||
#define STREAM_SCAN_WAIT 'W'
|
||||
#define STREAM_SCAN_DATA 'D'
|
||||
|
|
@ -73,9 +72,12 @@
|
|||
extern bool executorEarlyStop();
|
||||
|
||||
/* release memory of communication layer, just for LLT */
|
||||
// 此函数的功能是释放通信层的内存
|
||||
int gs_release_comm_memory()
|
||||
{
|
||||
// 构造一个通信上下文的对象，切换到全局通信内存上下文，用于管理通信内存。
|
||||
AutoContextSwitch commContext(g_instance.comm_cxt.comm_global_mem_cxt);
|
||||
// 调用函数gs_r_release_comm_memory来释放通信内存。
|
||||
gs_r_release_comm_memory();
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -85,27 +87,41 @@ int gs_release_comm_memory()
|
|||
*
|
||||
* @param[IN] key_s: stream key
|
||||
*/
|
||||
void gs_memory_init_entry(StreamSharedContext* sharedContext, int consumerNum, int producerNum)
|
||||
// 此函数用于初始化一个关于流和内存使用情况的哈希表条目
|
||||
void gs_memory_init_entry(StreamSharedContext *sharedContext, int consumerNum, int producerNum)
|
||||
{
|
||||
struct hash_entry* entry = NULL;
|
||||
struct hash_entry** poll_entrys = NULL;
|
||||
struct hash_entry*** quota_entrys = NULL;
|
||||
// 定义一个指向哈希表条目的指针
|
||||
struct hash_entry *entry = NULL;
|
||||
// 定义一个指向哈希表条目的指针的指针，用于存储与消费者相关的条目
|
||||
struct hash_entry **poll_entrys = NULL;
|
||||
// 定义一个指向哈希表条目的指针的指针的指针，用于存储与生产者相关的条目
|
||||
struct hash_entry ***quota_entrys = NULL;
|
||||
|
||||
poll_entrys = (struct hash_entry**)palloc(sizeof(struct hash_entry*) * consumerNum);
|
||||
quota_entrys = (struct hash_entry***)palloc(sizeof(struct hash_entry**) * consumerNum);
|
||||
|
||||
for (int i = 0; i < consumerNum; i++) {
|
||||
entry = (struct hash_entry*)palloc(sizeof(struct hash_entry));
|
||||
// 为每个消费者分配内存以存储哈希表条目指针
|
||||
poll_entrys = (struct hash_entry **)palloc(sizeof(struct hash_entry *) * consumerNum);
|
||||
// 为每个消费者分配内存以存储生产者相关的哈希表条目指针
|
||||
quota_entrys = (struct hash_entry ***)palloc(sizeof(struct hash_entry **) * consumerNum);
|
||||
// 遍历所有消费者
|
||||
for (int i = 0; i < consumerNum; i++)
|
||||
{
|
||||
// 为每个消费者分配一个哈希表条目，并初始化它
|
||||
entry = (struct hash_entry *)palloc(sizeof(struct hash_entry));
|
||||
(void)entry->_init();
|
||||
// 将当前消费者的哈希表条目存储在poll_entrys中
|
||||
poll_entrys[i] = entry;
|
||||
quota_entrys[i] = (struct hash_entry**)palloc(sizeof(struct hash_entry*) * producerNum);
|
||||
for (int j = 0; j < producerNum; j++) {
|
||||
entry = (struct hash_entry*)palloc(sizeof(struct hash_entry));
|
||||
// 为每个生产者分配内存以存储哈希表条目指针
|
||||
quota_entrys[i] = (struct hash_entry **)palloc(sizeof(struct hash_entry *) * producerNum);
|
||||
// 遍历所有的生产者
|
||||
for (int j = 0; j < producerNum; j++)
|
||||
{
|
||||
// 为每个生产者分配一个哈希表条目，并初始化它
|
||||
entry = (struct hash_entry *)palloc(sizeof(struct hash_entry));
|
||||
(void)entry->_init();
|
||||
// 将当前生产者的哈希表条目存储在quota_entrys中
|
||||
quota_entrys[i][j] = entry;
|
||||
}
|
||||
}
|
||||
|
||||
// 将poll_entrys和quota_entrys存储在共享上下文中，以便后续使用
|
||||
sharedContext->poll_entrys = poll_entrys;
|
||||
sharedContext->quota_entrys = quota_entrys;
|
||||
}
|
||||
|
|
@ -117,38 +133,51 @@ void gs_memory_init_entry(StreamSharedContext* sharedContext, int consumerNum, i
|
|||
* @param[IN] sharedContext: context for shared memory stream
|
||||
* @param[IN] nthChannel: destination consumer
|
||||
*/
|
||||
void gs_message_by_memory(StringInfo buf, StreamSharedContext* sharedContext, int nthChannel)
|
||||
// 此函数的作用是通过内存发送错误/通知消息，其中参数buf为错误或者通知的字符串，sharedContext为共享内存流上下文，nthChannel为目标消费者
|
||||
void gs_message_by_memory(StringInfo buf, StreamSharedContext *sharedContext, int nthChannel)
|
||||
{
|
||||
// 目标缓冲区，用于存储要发送的消息
|
||||
StringInfo buf_dst = NULL;
|
||||
struct hash_entry* entry = NULL;
|
||||
// 哈希表条目，用于管理共享内存流
|
||||
struct hash_entry *entry = NULL;
|
||||
|
||||
/* Copy Error/Notice messages to shared context. */
|
||||
// 将错误/通知消息复制到共享上下文中
|
||||
buf_dst = sharedContext->messages[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
|
||||
/*
|
||||
* If producer is waked up and shared buffer has been consumed while waiting,
|
||||
* it can continue to append data to its messages of sharedContext.
|
||||
*/
|
||||
// 如果生产者在等待期间被唤醒，并且共享缓冲区已被消耗，它可以继续将其数据追加到sharedContext的消息中
|
||||
entry = sharedContext->quota_entrys[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
while (buf_dst->len > 0) {
|
||||
// 当目标缓冲区中还有未处理的数据时
|
||||
while (buf_dst->len > 0)
|
||||
{
|
||||
// 等待一段时间，直到可以继续处理数据
|
||||
(void)entry->_timewait(SINGLE_WAITQUOTA);
|
||||
}
|
||||
// 将源缓冲区的数据追加到目标缓冲区中
|
||||
appendBinaryStringInfo(buf_dst, buf->data, buf->len);
|
||||
// 更新目标缓冲区的游标位置
|
||||
buf_dst->cursor = buf->cursor;
|
||||
|
||||
/* Send signal to dest consumer. */
|
||||
// 向目标消费者发送信号。
|
||||
entry = sharedContext->poll_entrys[nthChannel];
|
||||
// 发送信号通知目标消费者有新消息到达
|
||||
entry->_signal();
|
||||
|
||||
// 释放源缓冲区的数据内存
|
||||
pfree(buf->data);
|
||||
// 将源缓冲区的数据指针置为NULL，避免悬挂指针
|
||||
buf->data = NULL;
|
||||
}
|
||||
|
||||
void gs_memory_disconnect(StreamSharedContext* sharedContext, int nthChannel)
|
||||
// 此函数的作用是断开内存连接
|
||||
void gs_memory_disconnect(StreamSharedContext *sharedContext, int nthChannel)
|
||||
{
|
||||
struct hash_entry* entry = NULL;
|
||||
// 定义一个指向哈希表条目的指针
|
||||
struct hash_entry *entry = NULL;
|
||||
// 将指定通道的数据状态设置为连接错误
|
||||
sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = CONN_ERR;
|
||||
// 获取指定通道的轮询条目
|
||||
entry = sharedContext->poll_entrys[nthChannel];
|
||||
// 向轮询条目发送信号，通常用于通知其他进程或线程发生了某种事件或状态变化
|
||||
entry->_signal();
|
||||
}
|
||||
|
||||
|
|
@ -159,16 +188,27 @@ void gs_memory_disconnect(StreamSharedContext* sharedContext, int nthChannel)
|
|||
* @param[IN] sharedContext: context for shared memory stream
|
||||
* @param[IN] nthChannel: destination consumer
|
||||
*/
|
||||
bool gs_is_databuff_empty(StreamSharedContext* sharedContext, int nthChannel)
|
||||
// 此函数的作用是判断数据缓冲区是否为空，其中参数sharedContext为共享内存流上下文，nthChannel为目标消费者
|
||||
bool gs_is_databuff_empty(StreamSharedContext *sharedContext, int nthChannel)
|
||||
{
|
||||
if (sharedContext->vectorized) {
|
||||
VectorBatch* batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
if (batch->m_rows == 0) {
|
||||
// 判断是否启用了向量化处理
|
||||
if (sharedContext->vectorized)
|
||||
{
|
||||
// 获取指定通道和会话的共享批处理对象
|
||||
VectorBatch *batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
// 如果批处理的行数为0，则缓冲区为空
|
||||
if (batch->m_rows == 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
} else {
|
||||
TupleVector* tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
if (tupleVec->tuplePointer == 0) {
|
||||
}
|
||||
else
|
||||
{
|
||||
// 获取指定通道和会话的共享元组向量对象
|
||||
TupleVector *tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
// 如果元组指针为0，则缓冲区为空
|
||||
if (tupleVec->tuplePointer == 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
|
@ -185,94 +225,134 @@ bool gs_is_databuff_empty(StreamSharedContext* sharedContext, int nthChannel)
|
|||
* @param[IN] nthChannel: destination consumer
|
||||
* @param[IN] nthRow: the Nth row to be sent in batch
|
||||
*/
|
||||
// 此函数的作用是通过共享内存向本地消费者发送消息，参数tuple为要发送的元组，batchsrc为要发送的批次，sharedContext为共享内存流上下文，nthChannel为目标消费者，nthRow为批次中要发送的第n行
|
||||
void gs_memory_send(
|
||||
TupleTableSlot* tuple, VectorBatch* batchsrc, StreamSharedContext* sharedContext, int nthChannel, int nthRow)
|
||||
TupleTableSlot *tuple, VectorBatch *batchsrc, StreamSharedContext *sharedContext, int nthChannel, int nthRow)
|
||||
{
|
||||
VectorBatch* batch = NULL;
|
||||
TupleVector* tupleVec = NULL;
|
||||
// 定义一个批处理指针，用于存储批处理对象
|
||||
VectorBatch *batch = NULL;
|
||||
// 定义一个元组向量指针，用于存储元组向量对象
|
||||
TupleVector *tupleVec = NULL;
|
||||
// 定义一个布尔型变量，用于存储是否可以发送的状态
|
||||
bool ready_to_send = false;
|
||||
// 定义一个数据状态变量，用于存储数据的状态
|
||||
DataStatus dataStatus;
|
||||
struct hash_entry* entry = NULL;
|
||||
|
||||
// 定义一个哈希表条目指针，用于存储共享内存流上下文的哈希表条目
|
||||
struct hash_entry *entry = NULL;
|
||||
// 报告等待状态，将当前状态设置为等待刷新数据状态
|
||||
WaitState oldStatus = pgstat_report_waitstatus_comm(STATE_WAIT_FLUSH_DATA,
|
||||
u_sess->pgxc_cxt.PGXCNodeId,
|
||||
-1,
|
||||
u_sess->stream_cxt.producer_obj->getParentPlanNodeId(),
|
||||
global_node_definition ? global_node_definition->num_nodes : -1);
|
||||
|
||||
u_sess->pgxc_cxt.PGXCNodeId,
|
||||
-1,
|
||||
u_sess->stream_cxt.producer_obj->getParentPlanNodeId(),
|
||||
global_node_definition ? global_node_definition->num_nodes : -1);
|
||||
// 记录时间，开始发送数据
|
||||
StreamTimeSendStart(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
// 获取指定通道和会话的共享内存流上下文的哈希表条目
|
||||
entry = sharedContext->quota_entrys[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
for (;;) {
|
||||
/* Check for interrupt at the beginning of the loop. */
|
||||
// 进入无限循环，直到发送完成或发生中断等条件退出循环
|
||||
for (;;)
|
||||
{
|
||||
// 在循环开始处检查中断。如果发生中断，则立即退出循环
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* Check if we should early stop. */
|
||||
/* Quit if the connection close, especially in a early close case. */
|
||||
if (executorEarlyStop() || sharedContext->is_connect_end[nthChannel][u_sess->stream_cxt.smp_id]) {
|
||||
// 检查是否需要提前停止。如果连接关闭，特别是在提前关闭的情况下，则退出循环
|
||||
if (executorEarlyStop() || sharedContext->is_connect_end[nthChannel][u_sess->stream_cxt.smp_id])
|
||||
{
|
||||
// 恢复等待状态为原始状态
|
||||
(void)pgstat_report_waitstatus(oldStatus);
|
||||
return;
|
||||
}
|
||||
|
||||
// 获取指定通道和会话的数据状态
|
||||
dataStatus = sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
/* Break the loop if we find quota. */
|
||||
// 如果数据状态为DATA_EMPTY且（在__aarch64__架构下，数据缓冲区为空），或者数据状态为DATA_PREPARE，则跳出循环
|
||||
if ((dataStatus == DATA_EMPTY
|
||||
#ifdef __aarch64__
|
||||
&& gs_is_databuff_empty(sharedContext, nthChannel)
|
||||
#endif
|
||||
) ||
|
||||
dataStatus == DATA_PREPARE) {
|
||||
) ||
|
||||
dataStatus == DATA_PREPARE)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
// 记录时间，开始等待配额
|
||||
StreamTimeWaitQuotaStart(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
// 调用entry的_timewait方法，传入SINGLE_WAITQUOTA作为参数，等待配额
|
||||
(void)entry->_timewait(SINGLE_WAITQUOTA);
|
||||
// 记录时间，结束等待配额
|
||||
StreamTimeWaitQuotaEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
}
|
||||
|
||||
// 记录时间，开始复制数据
|
||||
StreamTimeCopyStart(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
/* Copy data to shared context. */
|
||||
if (sharedContext->vectorized) {
|
||||
// 将数据复制到共享上下文
|
||||
if (sharedContext->vectorized)
|
||||
{
|
||||
// 如果启用了向量化处理，则断言共享批处理对象不为空
|
||||
Assert(sharedContext->sharedBatches != NULL);
|
||||
// 获取指定通道和会话的共享批处理对象
|
||||
batch = sharedContext->sharedBatches[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
/* data copy */
|
||||
if (-1 == nthRow) {
|
||||
/* Do deep copy of all rows, for local roundrobin & local broadcast. */
|
||||
// 如果nthRow为-1，则对所有行进行深度复制，用于本地循环和本地广播
|
||||
if (-1 == nthRow)
|
||||
{
|
||||
// Assert批处理的行数为0，因为要进行所有行的深度复制
|
||||
Assert(batch->m_rows == 0);
|
||||
// 进行深度复制
|
||||
batch->Copy<true, false>(batchsrc);
|
||||
// 设置可以发送的状态为true
|
||||
ready_to_send = true;
|
||||
} else {
|
||||
}
|
||||
else
|
||||
{
|
||||
// 复制指定行的数据
|
||||
batch->CopyNth(batchsrc, nthRow);
|
||||
if (BatchMaxSize == batch->m_rows) {
|
||||
// 如果批处理的行数等于BatchMaxSize，则设置可以发送的状态为true
|
||||
if (BatchMaxSize == batch->m_rows)
|
||||
{
|
||||
ready_to_send = true;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
}
|
||||
else
|
||||
{
|
||||
// 如果未启用向量化处理，则断言共享元组向量对象不为空
|
||||
Assert(sharedContext->sharedTuples != NULL);
|
||||
// 获取指定通道和会话的共享元组向量对象
|
||||
tupleVec = sharedContext->sharedTuples[nthChannel][u_sess->stream_cxt.smp_id];
|
||||
// 获取元组指针
|
||||
int n = tupleVec->tuplePointer;
|
||||
// 复制元组到元组向量
|
||||
ExecCopySlot(tupleVec->tupleVector[n], tuple);
|
||||
// 元组指针加1
|
||||
tupleVec->tuplePointer++;
|
||||
if (TupleVectorMaxSize == tupleVec->tuplePointer) {
|
||||
// 如果元组指针等于TupleVectorMaxSize，则设置可以发送的状态为true
|
||||
if (TupleVectorMaxSize == tupleVec->tuplePointer)
|
||||
{
|
||||
ready_to_send = true;
|
||||
}
|
||||
}
|
||||
// 记录时间，结束复制数据
|
||||
StreamTimeCopyEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
|
||||
/* send the signal if copy finished */
|
||||
if (ready_to_send) {
|
||||
// 如果数据已经准备好发送，则执行以下代码块
|
||||
if (ready_to_send)
|
||||
{
|
||||
#ifdef __aarch64__
|
||||
// 在__aarch64__架构下，执行内存屏障操作，确保内存操作的正确顺序
|
||||
pg_memory_barrier();
|
||||
#endif
|
||||
/* set flag */
|
||||
// 设置数据状态为DATA_READY，表示数据已经准备好
|
||||
sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = DATA_READY;
|
||||
/* send signal */
|
||||
entry = sharedContext->poll_entrys[nthChannel];
|
||||
entry->_signal();
|
||||
} else {
|
||||
}
|
||||
else
|
||||
{
|
||||
// 如果数据还没有准备好，则将数据状态设置为DATA_PREPARE，表示数据正在准备中
|
||||
sharedContext->dataStatus[nthChannel][u_sess->stream_cxt.smp_id] = DATA_PREPARE;
|
||||
}
|
||||
// 记录时间，结束发送数据
|
||||
StreamTimeSendEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
|
||||
// 恢复等待状态为原始状态
|
||||
(void)pgstat_report_waitstatus(oldStatus);
|
||||
}
|
||||
|
||||
|
|
@ -282,17 +362,22 @@ void gs_memory_send(
|
|||
* @param[IN] node: stream state
|
||||
* @return bool: true -- found data
|
||||
*/
|
||||
// 此函数用于从流状态的缓冲区中获取一个元组，参数node为流状态
|
||||
FORCE_INLINE
|
||||
bool gs_return_tuple(StreamState* node)
|
||||
bool gs_return_tuple(StreamState *node)
|
||||
{
|
||||
TupleVector* tupleVec = node->tempTupleVec;
|
||||
|
||||
if (tupleVec->tuplePointer == 0) {
|
||||
// 获取流状态的临时元组向量
|
||||
TupleVector *tupleVec = node->tempTupleVec;
|
||||
// 如果元组指针为0，表示没有数据可返回
|
||||
if (tupleVec->tuplePointer == 0)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// 元组指针减1，因为我们要返回的是当前指针指向的元组
|
||||
tupleVec->tuplePointer--;
|
||||
// 获取当前元组指针的索引
|
||||
int n = tupleVec->tuplePointer;
|
||||
// 将结果元组槽设置为当前元组指针指向的元组
|
||||
node->ss.ps.ps_ResultTupleSlot = tupleVec->tupleVector[n];
|
||||
|
||||
return true;
|
||||
|
|
@ -305,53 +390,67 @@ bool gs_return_tuple(StreamState* node)
|
|||
* @param[IN] loc: data location
|
||||
* @return bool: true -- found data
|
||||
*/
|
||||
bool gs_consume_memory_data(StreamState* node, int loc)
|
||||
// 此函数的作用是从共享内存中消费本地生产者的数据，参数node为流状态，loc为数据位置
|
||||
bool gs_consume_memory_data(StreamState *node, int loc)
|
||||
{
|
||||
StreamSharedContext* sharedContext = node->sharedContext;
|
||||
|
||||
// 获取流状态的共享上下文
|
||||
StreamSharedContext *sharedContext = node->sharedContext;
|
||||
// 记录时间，开始网络工作时间拷贝
|
||||
NetWorkTimeCopyStart(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
/* Take data from the shared context. */
|
||||
if (sharedContext->vectorized) {
|
||||
VectorBatch* batchsrc = sharedContext->sharedBatches[u_sess->stream_cxt.smp_id][loc];
|
||||
VectorBatch* batchdst = ((VecStreamState*)node)->m_CurrentBatch;
|
||||
|
||||
if (batchsrc->m_rows == 0) {
|
||||
// 如果共享上下文已经向量化,从共享上下文中获取数据
|
||||
if (sharedContext->vectorized)
|
||||
{
|
||||
// 获取源批处理对象和目标批处理对象
|
||||
VectorBatch *batchsrc = sharedContext->sharedBatches[u_sess->stream_cxt.smp_id][loc];
|
||||
VectorBatch *batchdst = ((VecStreamState *)node)->m_CurrentBatch;
|
||||
// 如果源批处理的行数为0，表示没有数据可消费，返回false
|
||||
if (batchsrc->m_rows == 0)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// 将源批处理的数据复制到目标批处理，进行深复制，不重置源批处理
|
||||
batchdst->Copy<true, false>(batchsrc);
|
||||
|
||||
// 重置源批处理
|
||||
batchsrc->Reset();
|
||||
} else {
|
||||
TupleVector* tuplesrc = sharedContext->sharedTuples[u_sess->stream_cxt.smp_id][loc];
|
||||
TupleVector* tupledst = node->tempTupleVec;
|
||||
}
|
||||
else
|
||||
{
|
||||
// 如果共享上下文未向量化,获取源元组向量对象和目标元组向量对象
|
||||
TupleVector *tuplesrc = sharedContext->sharedTuples[u_sess->stream_cxt.smp_id][loc];
|
||||
TupleVector *tupledst = node->tempTupleVec;
|
||||
|
||||
if (tuplesrc->tuplePointer == 0) {
|
||||
// 如果源元组指针为0，表示没有数据可消费，返回false
|
||||
if (tuplesrc->tuplePointer == 0)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
for (int i = 0; i < tuplesrc->tuplePointer; i++) {
|
||||
// 将源元组向量的数据复制到目标元组向量
|
||||
for (int i = 0; i < tuplesrc->tuplePointer; i++)
|
||||
{
|
||||
(void)ExecCopySlot(tupledst->tupleVector[i], tuplesrc->tupleVector[i]);
|
||||
}
|
||||
|
||||
// 设置目标元组指针为源元组指针，重置源元组指针
|
||||
tupledst->tuplePointer = tuplesrc->tuplePointer;
|
||||
tuplesrc->tuplePointer = 0;
|
||||
// 返回元组
|
||||
(void)gs_return_tuple(node);
|
||||
}
|
||||
// 记录时间，结束网络工作时间拷贝
|
||||
NetWorkTimeCopyEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
|
||||
struct hash_entry* entry = NULL;
|
||||
struct hash_entry *entry = NULL;
|
||||
// 获取配额条目
|
||||
entry = sharedContext->quota_entrys[u_sess->stream_cxt.smp_id][loc];
|
||||
|
||||
// 如果编译在aarch64架构下，执行内存屏障操作，确保内存操作的正确顺序
|
||||
#ifdef __aarch64__
|
||||
pg_memory_barrier();
|
||||
#endif
|
||||
/* Reset flag */
|
||||
// 重置标志位
|
||||
sharedContext->dataStatus[u_sess->stream_cxt.smp_id][loc] = DATA_EMPTY;
|
||||
|
||||
/* send signal */
|
||||
// 发送信号
|
||||
entry->_signal();
|
||||
|
||||
// 更新扫描位置
|
||||
node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id] = loc;
|
||||
return true;
|
||||
}
|
||||
|
|
@ -364,97 +463,139 @@ bool gs_consume_memory_data(StreamState* node, int loc)
|
|||
* STREAM_SCAN_WAIT -- still need to poll to wait for data.
|
||||
* STREAM_SCAN_FINISH -- stream scan finished.
|
||||
*/
|
||||
char gs_find_memory_data(StreamState* node, int* waitnode_count)
|
||||
// 此函数的作用是从生产者状态扫描数据，参数node为流状态
|
||||
char gs_find_memory_data(StreamState *node, int *waitnode_count)
|
||||
{
|
||||
// 定义一个数据状态变量
|
||||
DataStatus dataStatus;
|
||||
// 定义一个字符串信息变量，初始值为NULL
|
||||
StringInfo buf = NULL;
|
||||
// 获取上次扫描的位置
|
||||
int scanLoc = node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id];
|
||||
// 定义一个计数器变量，初始值为上次扫描的位置
|
||||
int i = scanLoc;
|
||||
// 定义一个标志位，表示扫描是否完成，初始值为true
|
||||
bool finished = true;
|
||||
// 定义一个标志位，表示连接是否结束，初始值为false
|
||||
bool is_conn_end = false;
|
||||
// 定义一个计数器，用于统计需要等待的节点数量，初始值为0
|
||||
int waitnodeCount = 0;
|
||||
struct hash_entry* entry = NULL;
|
||||
// 定义一个哈希表条目指针，初始值为NULL
|
||||
struct hash_entry *entry = NULL;
|
||||
|
||||
/* Check if there is available data, and scan from last time location. */
|
||||
do {
|
||||
// 检查是否存在目标数据，并且从最后位置开始寻找
|
||||
do
|
||||
{
|
||||
i++;
|
||||
if (i == node->conn_count) {
|
||||
// 如果计数器等于连接数，则重置为0
|
||||
if (i == node->conn_count)
|
||||
{
|
||||
i = 0;
|
||||
}
|
||||
|
||||
/* Update scan location. */
|
||||
// 更新扫描位置
|
||||
node->sharedContext->scanLoc[u_sess->stream_cxt.smp_id] = i;
|
||||
// 获取当前位置的数据状态
|
||||
dataStatus = node->sharedContext->dataStatus[u_sess->stream_cxt.smp_id][i];
|
||||
// 获取当前位置的连接是否结束状态
|
||||
is_conn_end = node->sharedContext->is_connect_end[u_sess->stream_cxt.smp_id][i];
|
||||
|
||||
if (!is_conn_end) {
|
||||
if (!is_conn_end)
|
||||
{
|
||||
// 设置扫描完成标志位为false
|
||||
finished = false;
|
||||
// 需要等待的节点数量加1
|
||||
waitnodeCount++;
|
||||
}
|
||||
|
||||
/*
|
||||
* Firstly, we handle error or notice messages.
|
||||
* If an error occured, we should stop scan now.
|
||||
* If an notice occured, we can still receive data.
|
||||
*/
|
||||
// 首先，我们处理错误或通知消息。如果错误，我们应该立即停止。如果发生了通知，我们仍然可以接收数据
|
||||
// 从共享上下文中获取消息，这是一个字符串信息（StringInfo）结构，其中包含了消息的数据和长度等信息
|
||||
buf = node->sharedContext->messages[u_sess->stream_cxt.smp_id][i];
|
||||
if (buf->len > 0) {
|
||||
if (buf->cursor == 'E') {
|
||||
// 如果消息的长度大于0，即存在消息
|
||||
if (buf->len > 0)
|
||||
{
|
||||
// 如果消息的游标为'E'，表示这是一个错误消息
|
||||
if (buf->cursor == 'E')
|
||||
{
|
||||
// 调用函数处理流错误，参数为节点，错误消息的数据和长度
|
||||
HandleStreamError(node, buf->data, buf->len);
|
||||
// 返回一个标识，表示流扫描结束
|
||||
return STREAM_SCAN_FINISH;
|
||||
} else if (buf->cursor == 'N') {
|
||||
}
|
||||
// 如果消息的游标为'N'，表示这是一个通知消息
|
||||
else if (buf->cursor == 'N')
|
||||
{
|
||||
// 调用函数处理流通知，参数为节点，通知消息的数据和长度
|
||||
HandleStreamNotice(node, buf->data, buf->len);
|
||||
// 重置字符串信息，清空游标和数据
|
||||
resetStringInfo(buf);
|
||||
|
||||
/* After one notice message has handled, send signal and wake up the dest producer. */
|
||||
// 在处理完一个通知消息后，发送信号并唤醒目标生产者
|
||||
entry = node->sharedContext->quota_entrys[u_sess->stream_cxt.smp_id][i];
|
||||
// 获取配额条目，可能是为了记录或控制生产者的行为
|
||||
entry->_signal();
|
||||
|
||||
// 返回一个标识，表示流扫描需要等待
|
||||
return STREAM_SCAN_WAIT;
|
||||
}
|
||||
}
|
||||
|
||||
switch (dataStatus) {
|
||||
case DATA_EMPTY:
|
||||
break;
|
||||
|
||||
case DATA_PREPARE:
|
||||
/* Take the rest data away when the connection is end. */
|
||||
if (is_conn_end) {
|
||||
/* Return data if any. */
|
||||
if (gs_consume_memory_data(node, i)) {
|
||||
return STREAM_SCAN_DATA;
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case DATA_READY:
|
||||
if (gs_consume_memory_data(node, i)) {
|
||||
// 根据dataStatus的值选择执行的代码块
|
||||
switch (dataStatus)
|
||||
{
|
||||
// 如果dataStatus的值为DATA_EMPTY,不执行任何操作
|
||||
case DATA_EMPTY:
|
||||
break;
|
||||
// 如果dataStatus的值为DATA_PREPARE
|
||||
case DATA_PREPARE:
|
||||
// 当连接结束的时候带走其余的数据
|
||||
if (is_conn_end)
|
||||
{
|
||||
/* Return data if any. */
|
||||
// 如果有数据，通过调用gs_consume_memory_data函数来消耗数据
|
||||
if (gs_consume_memory_data(node, i))
|
||||
{
|
||||
// 返回STREAM_SCAN_DATA，表示成功从生产者找到数据
|
||||
return STREAM_SCAN_DATA;
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case CONN_ERR:
|
||||
ereport(ERROR,
|
||||
case DATA_READY:
|
||||
// 通过调用gs_consume_memory_data函数来消耗数据
|
||||
if (gs_consume_memory_data(node, i))
|
||||
{
|
||||
// 返回STREAM_SCAN_DATA，表示成功从生产者找到数据
|
||||
return STREAM_SCAN_DATA;
|
||||
}
|
||||
else
|
||||
{
|
||||
break;
|
||||
}
|
||||
// 如果dataStatus的值为CONN_ERR，生成一个错误报告
|
||||
case CONN_ERR:
|
||||
ereport(ERROR,
|
||||
(errcode(ERRCODE_STREAM_REMOTE_CLOSE_SOCKET),
|
||||
errmsg("Failed to read response from Local Stream Node,"
|
||||
" Detail: Node %s, Plan Node ID %u, SMP ID %d",
|
||||
errmsg("Failed to read response from Local Stream Node,"
|
||||
" Detail: Node %s, Plan Node ID %u, SMP ID %d",
|
||||
g_instance.attr.attr_common.PGXCNodeName,
|
||||
node->sharedContext->key_s.planNodeId,
|
||||
i)));
|
||||
break;
|
||||
// dataStatus is enum,
|
||||
default:
|
||||
break;
|
||||
break;
|
||||
// dataStatus is enum,
|
||||
default:
|
||||
break;
|
||||
}
|
||||
} while (i != scanLoc);
|
||||
|
||||
// 将waitnodeCount的值赋给指针waitnode_count所指向的变量，为了返回等待节点的数量
|
||||
*waitnode_count = waitnodeCount;
|
||||
|
||||
if (finished) {
|
||||
if (finished)
|
||||
{
|
||||
// 返回STREAM_SCAN_FINISH，表示流扫描完成
|
||||
return STREAM_SCAN_FINISH;
|
||||
} else {
|
||||
}
|
||||
else
|
||||
{
|
||||
// 返回STREAM_SCAN_WAIT，表示仍然需要轮询等待数据
|
||||
return STREAM_SCAN_WAIT;
|
||||
}
|
||||
}
|
||||
|
|
@ -466,51 +607,68 @@ char gs_find_memory_data(StreamState* node, int* waitnode_count)
|
|||
* @return bool: true -- successed to find data and need more data.
|
||||
* false -- all connection finished or recerive error.
|
||||
*/
|
||||
bool gs_memory_recv(StreamState* node)
|
||||
// 此函数的作用是从共享内存中接收本地流的数据，返回是否成功找到数据并需要更多数据，或者所有连接已完成或接收错误
|
||||
bool gs_memory_recv(StreamState *node)
|
||||
{
|
||||
// 存储操作结果的字符变量
|
||||
char result;
|
||||
struct hash_entry* entry = NULL;
|
||||
// 哈希表条目指针，初始化为NULL
|
||||
struct hash_entry *entry = NULL;
|
||||
// 获取流状态对应的哈希表条目
|
||||
entry = node->sharedContext->poll_entrys[u_sess->stream_cxt.smp_id];
|
||||
bool re = true;
|
||||
// 初始化等待节点数为0
|
||||
int waitnode_count = 0;
|
||||
|
||||
/* If there is already tuple in buffer, return the data at once. */
|
||||
if (!node->sharedContext->vectorized && gs_return_tuple(node)) {
|
||||
// 如果缓冲区中已有元组，则立即返回数据
|
||||
if (!node->sharedContext->vectorized && gs_return_tuple(node))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
for (;;) {
|
||||
/* Check for interrupt at the beginning of the loop. */
|
||||
for (;;)
|
||||
{
|
||||
// 检查是否有中断请求
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* Check if we can early stop now. */
|
||||
if (executorEarlyStop()) {
|
||||
// 检查是否可以提前结束循环
|
||||
if (executorEarlyStop())
|
||||
{
|
||||
re = false;
|
||||
break;
|
||||
}
|
||||
|
||||
/* Search all producers to find data. */
|
||||
// 搜索所有生产者以查找数据，同时更新等待节点数
|
||||
result = gs_find_memory_data(node, &waitnode_count);
|
||||
if (result == STREAM_SCAN_DATA) {
|
||||
if (result == STREAM_SCAN_DATA)
|
||||
{
|
||||
re = true;
|
||||
break;
|
||||
} else if (result == STREAM_SCAN_FINISH) {
|
||||
}
|
||||
// 如果所有连接已完成或接收错误
|
||||
else if (result == STREAM_SCAN_FINISH)
|
||||
{
|
||||
re = false;
|
||||
break;
|
||||
}
|
||||
|
||||
// 定义一个旧的等待状态阶段变量，初始化为PHASE_NONE，表示当前没有等待状态
|
||||
WaitStatePhase oldPhase = pgstat_report_waitstatus_phase(PHASE_NONE, true);
|
||||
// 定义一个旧的等待状态变量，通过调用pgstat_report_waitstatus_comm函数来初始化。
|
||||
// 该函数将等待状态设置为STATE_WAIT_NODE，表示当前正在等待节点响应。
|
||||
// 还将当前节点的ID、等待节点的数量、计划节点的ID以及全局节点定义的数量作为参数传递给该函数
|
||||
WaitState oldStatus = pgstat_report_waitstatus_comm(STATE_WAIT_NODE,
|
||||
u_sess->pgxc_cxt.PGXCNodeId,
|
||||
waitnode_count,
|
||||
node->sharedContext->key_s.planNodeId,
|
||||
global_node_definition ? global_node_definition->num_nodes : -1);
|
||||
u_sess->pgxc_cxt.PGXCNodeId,
|
||||
waitnode_count,
|
||||
node->sharedContext->key_s.planNodeId,
|
||||
global_node_definition ? global_node_definition->num_nodes : -1);
|
||||
|
||||
/* Poll to wait data from producers. */
|
||||
// 开始网络时间轮询，用于度量网络操作的耗时
|
||||
NetWorkTimePollStart(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
// 调用entry的_timewait方法，传入SINGLE_WAITQUOTA作为参数，用于等待生产者提供数据
|
||||
(void)entry->_timewait(SINGLE_WAITQUOTA);
|
||||
// 结束网络时间轮询
|
||||
NetWorkTimePollEnd(t_thrd.pgxc_cxt.GlobalNetInstr);
|
||||
|
||||
// 重置等待状态阶段和等待状态为旧的状态
|
||||
pgstat_reset_waitStatePhase(oldStatus, oldPhase);
|
||||
}
|
||||
|
||||
|
|
@ -523,16 +681,20 @@ bool gs_memory_recv(StreamState* node)
|
|||
* @param[IN] sharedContext: context for shared memory stream
|
||||
* @param[IN] connNum: producer connection number
|
||||
*/
|
||||
void gs_memory_send_finish(StreamSharedContext* sharedContext, int connNum)
|
||||
// 此函数用于通知所有相关的消费者没有更多数据可发送
|
||||
void gs_memory_send_finish(StreamSharedContext *sharedContext, int connNum)
|
||||
{
|
||||
struct hash_entry* entry = NULL;
|
||||
// 定义一个哈希表条目指针，初始化为NULL
|
||||
struct hash_entry *entry = NULL;
|
||||
|
||||
for (int i = 0; i < connNum; i++) {
|
||||
/* Set flags. */
|
||||
for (int i = 0; i < connNum; i++)
|
||||
{
|
||||
// 设置标志位，表示连接已经结束
|
||||
sharedContext->is_connect_end[i][u_sess->stream_cxt.smp_id] = true;
|
||||
|
||||
/* send signal */
|
||||
// 获取当前连接对应的哈希表条目
|
||||
entry = sharedContext->poll_entrys[i];
|
||||
// 调用哈希表条目的_signal方法，发送信号且通知等操作
|
||||
entry->_signal();
|
||||
}
|
||||
}
|
||||
|
|
@ -544,12 +706,15 @@ void gs_memory_send_finish(StreamSharedContext* sharedContext, int connNum)
|
|||
* @param[IN] connNum: producer connection number
|
||||
* @param[IN] smpId: producer smp id
|
||||
*/
|
||||
void gs_memory_close_conn(StreamSharedContext* sharedContext, int connNum, int consumerId)
|
||||
// 此函数用于设置与特定生产者的所有连接关闭
|
||||
void gs_memory_close_conn(StreamSharedContext *sharedContext, int connNum, int consumerId)
|
||||
{
|
||||
struct hash_entry* entry = NULL;
|
||||
// 定义一个哈希表条目指针，初始化为NULL
|
||||
struct hash_entry *entry = NULL;
|
||||
|
||||
for (int i = 0; i < connNum; i++) {
|
||||
/* Set flags. */
|
||||
for (int i = 0; i < connNum; i++)
|
||||
{
|
||||
// 设置标志位，表示与特定生产者的连接已结束
|
||||
sharedContext->is_connect_end[consumerId][i] = true;
|
||||
|
||||
/*
|
||||
|
|
@ -557,8 +722,7 @@ void gs_memory_close_conn(StreamSharedContext* sharedContext, int connNum, int c
|
|||
* in a query like "limit XXX", when consumer don't need data anymore,
|
||||
* but the producers haven't send all data yet.
|
||||
*/
|
||||
entry = sharedContext->quota_entrys[consumerId][i];
|
||||
entry->_signal();
|
||||
entry = sharedContext->quota_entrys[consumerId][i]; // 获取当前连接对应的哈希表条目
|
||||
entry->_signal(); // 调用哈希表条目的_signal方法，发送信号且通知等操作
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue