hcq/openGauss-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
v6.0.0
openGauss-server/contrib/ndpplugin/ndpam.cpp

1131 lines
36 KiB
C++
Raw Permalink Blame History

/* -------------------------------------------------------------------------
* ndpam.cpp
* Routines to handle ndp page
*
* Portions Copyright (c) 2022 Huawei Technologies Co.,Ltd.
*
* IDENTIFICATION
* contrib/ndpplugin/ndpam.cpp
*
* -------------------------------------------------------------------------
*/
#include "access/csnlog.h"
#include "access/slru.h"
#include "executor/node/nodeAgg.h"
#include "component/rpc/rpc.h"
#include "storage/smgr/segment_internal.h"
#include "ddes/dms/ss_transaction.h"
#include "algorithm"
#include "storage/dss/fio_dss.h"
#include "storage/smgr/segment.h"
#include "ndpnodes.h"
#include "ndpam.h"
#define NDP_PAGE_QUEUE_SIZE (1u << 10)
#define NDP_NORMAL_QUEUE_SIZE (1u << 12)
#define ClogCtl(n) (&t_thrd.shemem_ptr_cxt.ClogCtl[CBufHashPartition(n)])
#define CsnlogCtl(n) (&t_thrd.shemem_ptr_cxt.CsnlogCtlPtr[CSNBufHashPartition(n)])
#define CSNLOG_XACTS_PER_PAGE (BLCKSZ / sizeof(CommitSeqNo))
#define CSN_LWLOCK_ACQUIRE(pageno, lockmode) ((void)LWLockAcquire(CSNBufMappingPartitionLock(pageno), lockmode))
#define CSN_LWLOCK_RELEASE(pageno) (LWLockRelease(CSNBufMappingPartitionLock(pageno)))
#define TransactionIdToCSNPage(xid) ((xid) / (TransactionId)CSNLOG_XACTS_PER_PAGE)
constexpr int RPC_FAILED_LIMIT = 3;
constexpr int SEND_FAILED_LIMIT = 10;
inline static void SegLogicPageIdToExtentOffset(BlockNumber logicId, uint32* offset)
{
if (logicId < EXT_SIZE_8_TOTAL_PAGES) {
*offset = logicId % EXT_SIZE_8;
} else if (logicId < EXT_SIZE_128_TOTAL_PAGES) {
logicId -= EXT_SIZE_8_TOTAL_PAGES;
*offset = logicId % EXT_SIZE_128;
} else if (logicId < EXT_SIZE_1024_TOTAL_PAGES) {
logicId -= EXT_SIZE_128_TOTAL_PAGES;
*offset = logicId % EXT_SIZE_1024;
} else {
logicId -= EXT_SIZE_1024_TOTAL_PAGES;
*offset = logicId % EXT_SIZE_8192;
}
}
static void md_get_physical_info(Relation rel, ForkNumber forknum, BlockNumber blocknum,
int *handle, off_t *offset)
{
SMgrRelation reln = rel->rd_smgr;
MdfdVec *v = NULL;
v = _mdfd_getseg(reln, forknum, blocknum, false, EXTENSION_FAIL);
Assert(v != NULL);
*handle = FileFd(v->mdfd_vfd);
*offset = DF_OFFSET_TO_SLICE_OFFSET(((off_t)blocknum) * BLCKSZ);
}
void md_get_pageinfo(NdpScanDesc ndpScan, BlockNumber page, CephObject *object, char *ip,
BlockNumber& end, uint32& phyStartBlockNum)
{
// step1:get object and ip.
int handle;
off_t offset;
HeapScanDesc scan = (HeapScanDesc)ndpScan->scan;
md_get_physical_info(scan->rs_base.rs_rd, 0, page, &handle, &offset);
#ifdef GlobalCache
dss_get_addr(handle, offset, object->pool, object->image, ip, &object->objId, &object->objOffset);
#endif
// step2:slice pages, get end and phyStartBlockNum.
BlockNumber start = ndpScan->handledBlock;
#ifdef GlobalCache
Assert(object->objOffset >= 0);
end = Min((uint64_t)ndpScan->nBlock, (uint64_t)(start + PAGE_NUM_PER_AU - object->objOffset/BLCKSZ));
#else
end = Min((uint64_t)ndpScan->nBlock, (uint64_t)(start + PAGE_NUM_PER_AU));
#endif
phyStartBlockNum = NDPMERGEBIT(start, 0);
}
static void seg_get_physical_info(Relation rel, ForkNumber forknum, BlockNumber blocknum,
SegPageLocation &loc, int *handle, off_t *offset)
{
SMgrRelation reln = rel->rd_smgr;
loc = seg_get_physical_location(rel->rd_node, MAIN_FORKNUM, blocknum);
SegmentCheck(loc.blocknum != InvalidBlockNumber);
LockSegmentHeadPartition(reln->seg_space->spcNode, reln->seg_space->dbNode,
reln->seg_desc[forknum]->head_blocknum, LW_SHARED);
SegSpace *spc = reln->seg_space;
RelFileNode relNode = {.spcNode = spc->spcNode,
.dbNode = spc->dbNode,
.relNode = EXTENT_SIZE_TO_TYPE(loc.extent_size),
.bucketNode = SegmentBktId,
.opt = 0};
int egid = EXTENT_TYPE_TO_GROUPID(relNode.relNode);
SegExtentGroup *seg = &spc->extent_group[egid][forknum];
off_t beginoff = ((off_t)loc.blocknum) * BLCKSZ;
int sliceno = DF_OFFSET_TO_SLICENO(beginoff);
SegPhysicalFile spf = df_get_physical_file(seg->segfile, sliceno, loc.blocknum);
*handle = spf.fd;
*offset = DF_OFFSET_TO_SLICE_OFFSET(beginoff);
UnlockSegmentHeadPartition(reln->seg_space->spcNode, reln->seg_space->dbNode,
reln->seg_desc[forknum]->head_blocknum);
}
void seg_get_pageinfo(NdpScanDesc ndpScan, BlockNumber page, CephObject *object, char *ip,
BlockNumber& end, uint32& phyStartBlockNum)
{
// step1:get object and ip.
int handle;
off_t offset;
SegPageLocation loc;
HeapScanDesc scan = (HeapScanDesc)ndpScan->scan;
seg_get_physical_info(scan->rs_base.rs_rd, 0, page, loc, &handle, &offset);
#ifdef GlobalCache
dss_get_addr(handle, offset, object->pool, object->image, ip, &object->objId, &object->objOffset);
#endif
// step2:slice pages, get end and phyStartBlockNum.
uint32 extentOff;
BlockNumber start = ndpScan->handledBlock;
SegLogicPageIdToExtentOffset(ndpScan->handledBlock, &extentOff);
uint64_t blks = Min(loc.extent_size - extentOff,
Min(uint32(PAGE_NUM_PER_AU), uint32(ndpScan->nBlock - ndpScan->handledBlock)));
#ifdef GlobalCache
Assert(object->objOffset >= 0);
end = start + Min(blks, (uint64_t)(PAGE_NUM_PER_AU - object->objOffset/BLCKSZ));
#else
end = start + blks;
#endif
phyStartBlockNum = NDPMERGEBIT(loc.blocknum, EXTENT_SIZE_TO_GROUPID(loc.extent_size) + 1);
}
void pm_get_pageinfo(NdpScanDesc ndpScan, BlockNumber page, CephObject *object, char *ip,
BlockNumber& end, uint32& phyStartBlockNum)
{
HeapScanDesc scan = (HeapScanDesc)ndpScan->scan;
int which = scan->rs_base.rs_rd->rd_smgr->smgr_which;
FileType filetype = which == 0 ? MDFILE : (which == 2 ? SEGFILE : INVALIDFILE);
Assert(filetype != INVALIDFILE);
PAGEMETHOD[filetype].get_pageinfo(ndpScan, page, object, ip, end, phyStartBlockNum);
uint64 curAUAligned = ndpScan->handledBlock / PARALLEL_SCAN_GAP_AU_ALIGNED;
if (scan->dop > 1 && end > (curAUAligned + 1) * PARALLEL_SCAN_GAP_AU_ALIGNED) {
end = (curAUAligned + 1) * PARALLEL_SCAN_GAP_AU_ALIGNED;
}
}
void CopyCLog(int64 pageno, char *pageBuffer)
{
int slotno;
errno_t rc = EOK;
/* lock is acquired by SimpleLruReadPage_ReadOnly */
slotno = SimpleLruReadPage_ReadOnly(ClogCtl(pageno), pageno, FirstNormalTransactionId);
rc = memcpy_s(pageBuffer, BLCKSZ, ClogCtl(pageno)->shared->page_buffer[slotno], BLCKSZ);
securec_check(rc, "", "");
LWLockRelease(ClogCtl(pageno)->shared->control_lock);
}
void CopyCSNLog(int64 pageno, char *pageBuffer)
{
int slotno;
errno_t rc = EOK;
CSN_LWLOCK_ACQUIRE(pageno, LW_SHARED);
slotno = SimpleLruReadPage_ReadOnly_Locked(CsnlogCtl(pageno), pageno, FirstNormalTransactionId);
rc = memcpy_s(pageBuffer, BLCKSZ, CsnlogCtl(pageno)->shared->page_buffer[slotno], BLCKSZ);
securec_check(rc, "", "");
CSN_LWLOCK_RELEASE(pageno);
}
int NdpIoSlot::SetReq(RelFileNode& node, uint16 taskId, uint16 tableId, AuInfo& auinfo)
{
int bitCount = 0;
reqMsg.data = &req;
reqMsg.len = sizeof(NdpIORequest);
respMsg.data = nullptr;
respMsg.len = 0;
req.taskId = taskId;
req.tableId = tableId;
req.auInfos[0].phyStartBlockNum = auinfo.phyStartBlockNum;
req.auInfos[0].pageNum = auinfo.pageNum;
#ifdef GlobalCache
errno_t rc2 = memcpy_s(&req.auInfos[0].object, sizeof(CephObject), &auinfo.object, sizeof(CephObject));
securec_check(rc2, "", "");
#endif
errno_t rc = memset_s(req.pageMap, BITMAP_SIZE_PER_AU_BYTE, 0, BITMAP_SIZE_PER_AU_BYTE);
securec_check(rc, "", "");
for (uint32 i = startBlockNum, offset = 0; i != startBlockNum + auinfo.pageNum; ++i, ++offset) {
bool cached = IsPageHitBufferPool(node, MAIN_FORKNUM, i);
if (!cached) {
NDPSETBIT(req.pageMap, offset);
++bitCount;
}
}
if (SS_STANDBY_MODE) {
SSIsPageHitDms(node, startBlockNum, auinfo.pageNum, req.pageMap, &bitCount);
}
return bitCount;
}
NdpRetCode NdpIoSlot::SetResp(int pageNum)
{
#ifdef ENABLE_SSL
respMsg.len = 0;
respMsg.data = nullptr;
return NdpRetCode::NDP_OK;
#else
respMsg.len = DSS_DEFAULT_AU_SIZE;
if (g_ndp_instance.pageContext->Dequeue(respMsg.data)) {
resp = reinterpret_cast<NdpIOResponse*>(respMsg.data);
return NdpRetCode::NDP_OK;
}
return NdpRetCode::ALLOC_RESPONSE_MEMORY_FAILED;
#endif
}
NdpRetCode NdpIoSlot::GetResp(NdpPageHeader& pages, int& pageNum, BlockNumber& start, uint64*& map)
{
start = startBlockNum;
map = nullptr;
pages = nullptr;
pageNum = 0;
if (respRet != NdpRetCode::NDP_OK) {
return respRet;
}
auto rpcResp = reinterpret_cast<NdpIOResponse*>(respMsg.data);
if (rpcResp == nullptr) {
return NdpRetCode::NDP_RETURN_FAILED;
}
#ifdef ENABLE_SSL
resp = reinterpret_cast<NdpIOResponse*>(respMsg.data);
#endif
if (rpcResp->status != 0) {
return NdpRetCode::NDP_RETURN_STATUS_ERROR;
}
map = rpcResp->pageMap;
pageNum = rpcResp->ndpPageNums;
if (pageNum) {
pages = (NdpPageHeader)((char*)rpcResp + sizeof(NdpIOResponse));
}
return NdpRetCode::NDP_OK;
}
void NdpIoSlot::FreeResp()
{
#ifdef ENABLE_SSL
if (resp != nullptr) {
free(resp);
resp = nullptr;
return;
}
if (respMsg.data) {
free(respMsg.data);
respMsg.data = nullptr;
return;
}
#else
bool enqueued;
Assert(g_ndp_instance.pageContext);
if (resp) {
enqueued = g_ndp_instance.pageContext->Enqueue(resp);
if (!enqueued) {
ereport(WARNING, (errmsg("try enqueue slot memory to queue failed")));
}
resp = nullptr;
return;
}
if (respMsg.data) {
enqueued = g_ndp_instance.pageContext->Enqueue(respMsg.data);
if (!enqueued) {
ereport(WARNING, (errmsg("try enqueue slot memory to queue failed")));
}
respMsg.data = nullptr;
return;
}
#endif
}
NdpRetCode NdpScanDescData::Init(ScanState* sstate, TableScanDesc sscan)
{
curIO = nullptr; // necessary, because rescan may be before get next
// free everything in destructor if alloc failed
#ifdef NDP_ASYNC_RPC
pg_atomic_init_u32(&reqCount, 0);
pg_atomic_init_u32(&respCount, 0);
respIO = new MpmcBoundedQueue<NdpIoSlot*>(NDP_PAGE_QUEUE_SIZE);
if (respIO == nullptr) {
ereport(WARNING, (errmsg("Alloc NpdPage Queue failed, size %d", NDP_PAGE_QUEUE_SIZE)));
return NdpRetCode::ALLOC_MQ_FAILED;
}
normalPagesId = new MpmcBoundedQueue<int>(NDP_NORMAL_QUEUE_SIZE);
if (normalPagesId == nullptr) {
ereport(WARNING, (errmsg("Alloc Normal Queue failed, size %d", NDP_NORMAL_QUEUE_SIZE)));
return NdpRetCode::ALLOC_MQ_FAILED;
}
#endif
memCtx = AllocSetContextCreate(CurrentMemoryContext, "ThreadNdpScanContext", 0, (4u << 10), (4u << 10));
if (!memCtx) {
ereport(WARNING, (errmsg("Create ThreadNdpScanContext failed!")));
return NdpRetCode::ALLOC_MC_FAILED;
}
#ifdef FAULT_INJECT
if ((rand() % PERCENTAGE_DIV) < PERCENTAGE) {
ereport(WARNING, (errmsg("Fault inject -- Create ThreadNdpScanContext failed")));
return NdpRetCode::ALLOC_MC_FAILED;
}
#endif
cond = (NdpScanCondition*)sstate->ps.plan->ndp_pushdown_condition;
scan = sscan;
scanState = sstate;
aggState = NULL;
aggSlot = NULL;
return NdpRetCode::NDP_OK;
}
NdpScanDescData::~NdpScanDescData()
{
// wait all callback return
while (pg_atomic_read_u32(&reqCount) != pg_atomic_read_u32(&respCount)) {
pg_usleep(NDP_RPC_WAIT_USEC);
}
#ifdef NDP_ASYNC_RPC
if (respIO) {
NdpIoSlot* tmpIO = nullptr;
while (respIO->Dequeue(tmpIO)) {
delete tmpIO;
}
delete respIO;
}
if (normalPagesId) {
delete normalPagesId;
}
#endif
if (memCtx) {
MemoryContextDelete(memCtx);
}
}
void NdpScanDescData::Reset()
{
handledBlock = 0;
curLinesNum = 0;
nextLineIndex = 0;
#ifdef NDP_ASYNC_RPC
// add timestamp if we don't want to wait, and remember release all while deleting.
while (pg_atomic_read_u32(&respCount) < pg_atomic_read_u32(&reqCount)) {
pg_usleep(NDP_RPC_WAIT_USEC);
}
#endif
FreeCurSlot();
#ifdef NDP_ASYNC_RPC
NdpIoSlot* tmpIO = nullptr;
int tmpId;
while (respIO->Dequeue(tmpIO)) {
delete tmpIO;
}
while (normalPagesId->Dequeue(tmpId));
#else
normalPagesNum = 0;
#endif
MemoryContextReset(memCtx);
}
void NdpScanDescData::AddToNormal(uint32 start, uint32 end)
{
end = end > nBlock ? nBlock : end;
start = start > nBlock ? nBlock : start;
for (uint32 i = start; i < end; ++i) {
AddToNormal(i);
}
}
// return true if set current io slot;
bool NdpScanDescData::HandleSlot(NdpIoSlot* slot)
{
uint32 start;
uint64* pageMap;
NdpPageHeader pages;
int pageNum;
NdpRetCode ret = slot->GetResp(pages, pageNum, start, pageMap);
if (ret != NdpRetCode::NDP_OK) {
++failedIoN;
}
if (start > nBlock) {
ereport(ERROR, (errmsg("can not happen start %u is cross the border.", start)));
return false;
}
uint32 end = start + slot->GetPushDownPageNum();
if (pageMap == nullptr) { // slot is invalid, add to normal list
AddToNormal(start, end);
return false;
}
// put unhandled page to normal queue
int count = 0;
end = end > nBlock ? nBlock : end;
start = start > nBlock ? nBlock : start;
for (uint32 offset = start, i = 0; offset < end; ++i, ++offset) {
int flag = NDPGETBIT(pageMap, i);
if (!flag) {
AddToNormal(offset);
count++;
}
}
sendBackPageN += count;
if (scanState->ps.instrument) {
scanState->ps.instrument->ndp_sendback_page += count;
}
if (pages) {
curIO = slot;
curNdpPages = pages;
curNdpPagesNum = pageNum;
nextNdpPageIndex = 0;
if (pages->pd_flags == NDP_FILTERED_PAGE) {
ndpPageScanN += pageNum;
} else {
ndpPageAggN += pageNum;
}
if (scanState->ps.instrument) {
scanState->ps.instrument->ndp_handled += pageNum;
}
return true;
} else {
return false;
}
}
#ifdef NDP_ASYNC_RPC
bool NdpScanDescData::GetNextSlot(void)
{
NdpIoSlot* slot = nullptr;
if (respIO->Dequeue(slot)) {
if (HandleSlot(slot)) {
return true;
} else {
delete slot;
return false;
}
}
return false;
}
#endif
#ifdef NDP_ASYNC_RPC
void NdpIoSlotCallDone(RpcStatus status, void *arg)
{
NdpIoSlot* cbArg = reinterpret_cast<NdpIoSlot*>(arg);
if (!cbArg) {
return;
}
NdpScanDesc ndpScan = reinterpret_cast<NdpScanDesc>(cbArg->GetPriv());
if (ndpScan == nullptr) {
return;
}
#ifdef FAULT_INJECT
if ((rand() % PERCENTAGE_DIV) < PERCENTAGE) {
status = RPC_ERROR;
}
#endif
if (status != RPC_OK) {
cbArg->SetRespRet(NdpRetCode::RPC_IO_CALLBACK_ERROR);
cbArg->SetRPCRet(status);
}
while (!ndpScan->respIO->Enqueue(cbArg)) {
pg_usleep(NDP_RPC_WAIT_USEC);
}
pg_atomic_add_fetch_u32(&ndpScan->respCount, 1);
}
#endif
bool NdpScanChannel::Init(uint32 id, char* ip, uint32 tableN)
{
rpcId = id;
errno_t rc = strcpy_s(rpcIp, NDP_RPC_IP_LEN, ip);
securec_check(rc, "", "");
status = NdpScanChannelStatus::UNCONNECTED;
pthread_mutex_init(&mutex, nullptr);
rpcClient = 0;
queryId = 0;
tableNum = tableN;
tableMgr = New(CurrentMemoryContext) NdpTableMgr[tableNum]();
if (tableMgr == nullptr) {
return false;
}
connFailed = 0;
cmdFailed = 0;
return true;
}
NdpRetCode NdpScanChannel::SendRequest(NdpIoSlot* req, NdpScanDesc ndpScan)
{
if (status == NdpScanChannelStatus::CLOSED) {
return NdpRetCode::CONNECT_UNUSABLE;
}
if (req->GetReqTableId() >= tableNum) {
ereport(WARNING, (errmsg("table id %u should be littler then %u.", req->GetReqTableId(), tableNum)));
return NdpRetCode::TABLE_ID_INVALID;
}
if (status == NdpScanChannelStatus::QUERYSENT) {
return SendReq(req, ndpScan);
}
if (pthread_mutex_trylock(&mutex) != 0) {
return NdpRetCode::CONNECT_UNUSABLE;
}
NdpRetCode ret = NdpRetCode::NDP_OK;
switch (status) {
case NdpScanChannelStatus::UNCONNECTED: {
// free old connection
DisconnectRpc();
// call rpc connect
RpcStatus connectRet = RpcClientConnect(rpcIp, u_sess->ndp_cxt.ndp_port, rpcClient);
if (SECUREC_UNLIKELY(connectRet != RPC_OK)) {
++connFailed;
DisconnectRpc();
if (connFailed >= RPC_FAILED_LIMIT) {
status = NdpScanChannelStatus::CLOSED;
}
ereport(LOG, (errmsg("rpc connect (count:%d) failed, ip:port %s:%d. rpc status: %d",
connFailed, rpcIp, u_sess->ndp_cxt.ndp_port, connectRet)));
ret = NdpRetCode::CONNECT_FAILED;
break;
}
status = NdpScanChannelStatus::CONNECTED;
}
case NdpScanChannelStatus::CONNECTED: {
PG_TRY();
{
ret = SendQuery(ndpScan);
}
PG_CATCH();
{
ereport(WARNING, (errmsg("send query failed, it is possible a palloc failed.")));
pthread_mutex_unlock(&mutex);
PG_RE_THROW();
}
PG_END_TRY();
if (SECUREC_UNLIKELY(ret != NdpRetCode::NDP_OK)) {
++cmdFailed;
if (cmdFailed >= RPC_FAILED_LIMIT) {
status = NdpScanChannelStatus::CLOSED;
}
break;
}
status = NdpScanChannelStatus::QUERYSENT;
}
case NdpScanChannelStatus::QUERYSENT: {
pthread_mutex_unlock(&mutex);
return SendReq(req, ndpScan);
}
case NdpScanChannelStatus::CLOSED:
default: {
ret = NdpRetCode::CONNECT_UNUSABLE;
break;
}
}
pthread_mutex_unlock(&mutex);
return ret;
}
NdpRetCode NdpScanChannel::SendEnd()
{
NdpRetCode retCode = NdpRetCode::NDP_OK;
if (status != NdpScanChannelStatus::QUERYSENT) {
/* If all pages are cached in BufferPool, the channel might be UNCONNECTED */
ereport(LOG, (errmsg("channel %s is not QUERY_SENT.", rpcIp)));
return NdpRetCode::CONNECT_UNUSABLE;
}
pthread_mutex_lock(&mutex);
if (status == NdpScanChannelStatus::QUERYSENT) {
NdpAdminRequest req;
NdpAdminResponse resp;
req.head.command = NDP_TERMINATE;
req.head.size = sizeof(NdpAdminRequest);
req.taskId = queryId;
req.tableId = 0;
NdpRetCode ret = SendAdminReq(&req, &resp, sizeof(resp.ret));
if (ret != NdpRetCode::NDP_OK) {
retCode = NdpRetCode::RPC_ADMIN_SEND_TERMINATE_FAILED;
} else {
status = NdpScanChannelStatus::CLOSED;
}
}
pthread_mutex_unlock(&mutex);
return retCode;
}
NdpRetCode NdpScanChannel::SendReq(NdpIoSlot* req, NdpScanDesc ndpScan)
{
NdpTableMgr* mgr = &tableMgr[req->GetReqTableId()];
if (mgr->status == NdpTableStatus::CONSTRUCTFAIL) {
return NdpRetCode::NDP_CONSTRUCT_FAILED;
}
if (mgr->status == NdpTableStatus::STATESENT) {
return SendIo(req, ndpScan);
}
if (mgr->cmdNdpFailed >= RPC_FAILED_LIMIT) {
return NdpRetCode::RPC_ADMIN_SEND_FAIL;
}
if (pthread_mutex_trylock(&mgr->mutex) != 0) {
return NdpRetCode::TABLE_MGR_UNUSABLE;
}
NdpRetCode ret;
PG_TRY();
{
ret = SendAdmin(mgr, req, ndpScan);
}
PG_CATCH();
{
ereport(WARNING, (errmsg("send failed, it is possible a palloc failed.")));
ret = NdpRetCode::NDP_ERROR;
}
PG_END_TRY();
pthread_mutex_unlock(&mgr->mutex);
if (mgr->status == NdpTableStatus::STATESENT) {
return SendIo(req, ndpScan);
}
return ret;
}
NdpRetCode NdpScanChannel::SendAdmin(NdpTableMgr* mgr, NdpIoSlot* req, NdpScanDesc ndpScan)
{
NdpRetCode ret = NdpRetCode::NDP_OK;
switch (mgr->status) {
case NdpTableStatus::INITIAL: {
ret = SendPlan(ndpScan);
if (ret != NdpRetCode::NDP_OK) {
mgr->status = (ret == NdpRetCode::NDP_CONSTRUCT_FAILED) ?
NdpTableStatus::CONSTRUCTFAIL : mgr->status;
break;
}
mgr->status = NdpTableStatus::PLANSENT;
}
case NdpTableStatus::PLANSENT: {
ret = SendState(ndpScan);
if (ret != NdpRetCode::NDP_OK) {
mgr->status = (ret == NdpRetCode::NDP_CONSTRUCT_FAILED) ?
NdpTableStatus::CONSTRUCTFAIL : mgr->status;
break;
}
mgr->status = NdpTableStatus::STATESENT;
}
case NdpTableStatus::STATESENT: {
ret = NdpRetCode::NDP_OK;
break;
}
default: {
ret = NdpRetCode::NDP_ERROR;
break;
}
}
if (ret != NdpRetCode::NDP_OK) {
++mgr->cmdNdpFailed;
}
return ret;
}
NdpRetCode NdpScanChannel::SendQuery(NdpScanDesc ndpScan)
{
NdpAdminRequest *v = ConstructVersion();
NdpAdminResponse resp;
NdpRetCode versionRet = SendAdminReq(v, &resp, sizeof(NdpAdminResponse));
pfree(v);
if (versionRet != NdpRetCode::NDP_OK) {
DisconnectRpc();
status = NdpScanChannelStatus::CLOSED;
ereport(LOG, (errmsg("send version admin (count:%d) request failed, ip:port %s:%d.",
cmdFailed, rpcIp, u_sess->ndp_cxt.ndp_port)));
return NdpRetCode::RPC_ADMIN_SEND_VERSION_FAILED;
}
NdpAdminRequest *query;
resp = {};
query = ConstructQuery(ndpScan);
NdpRetCode queryRet = SendAdminReq(query, &resp, sizeof(NdpAdminResponse));
pfree(query);
if (queryRet != NdpRetCode::NDP_OK) {
ereport(LOG, (errmsg("send admin (count:%d) request failed, ip:port %s:%d.",
cmdFailed, rpcIp, u_sess->ndp_cxt.ndp_port)));
return NdpRetCode::RPC_ADMIN_SEND_CTX_FAILED;
}
queryId = (uint16)resp.queryId;
return NdpRetCode::NDP_OK;
}
NdpRetCode NdpScanChannel::SendPlan(NdpScanDesc ndpScan)
{
NdpAdminRequest* planReq = ConstructPlanReq(ndpScan);
if (planReq == nullptr) {
return NdpRetCode::NDP_CONSTRUCT_FAILED;
}
NdpAdminResponse resp;
NdpRetCode ret = SendAdminReq(planReq, &resp, sizeof(resp.ret));
pfree(planReq);
if (ret != NdpRetCode::NDP_OK) {
ret = NdpRetCode::RPC_ADMIN_SEND_PLAN_FAILED;
}
return ret;
}
NdpRetCode NdpScanChannel::SendAdminReq(NdpAdminRequest* req, NdpAdminResponse* resp, size_t size)
{
RpcStatus status = RpcSendAdminReq(req, resp, size, rpcClient);
if (status != RPC_OK) {
ereport(LOG, (errmsg("RpcSendAdminReq failed. CMD code:%d, Rpc code: %d", req->head.command, status)));
return NdpRetCode::RPC_ADMIN_SEND_FAIL;
} else if (resp->ret != NDP_OK) {
ereport(LOG, (errmsg("AdminReq handle failed.")));
return NdpRetCode::NDP_ERROR;
}
return NdpRetCode::NDP_OK;
}
NdpRetCode NdpScanChannel::SendState(NdpScanDesc ndpScan)
{
NdpAdminRequest* state = ConstructPlanState(ndpScan);
if (state == nullptr) {
return NdpRetCode::NDP_CONSTRUCT_FAILED;
}
NdpAdminResponse resp;
NdpRetCode ret = SendAdminReq(state, &resp, sizeof(resp.ret));
pfree(state);
if (ret != NdpRetCode::NDP_OK) {
ret = NdpRetCode::RPC_ADMIN_SEND_STATE_FAILED;
}
return ret;
}
NdpRetCode NdpScanChannel::SendIo(NdpIoSlot* req, NdpScanDesc ndpScan)
{
req->SetReq(queryId);
NdpTableMgr* mgr = &tableMgr[req->GetReqTableId()];
if (mgr->ioFailed >= SEND_FAILED_LIMIT) {
return NdpRetCode::RPC_IO_SEND_FAILED;
}
#ifdef NDP_ASYNC_RPC
RpcCallDone callDone = {.cb = &NdpIoSlotCallDone, .arg = (void*)req };
pg_atomic_add_fetch_u32(&ndpScan->reqCount, 1); // before send
RpcStatus ret = RpcSendIOReq(req->GetReqMsg(), req->GetRespMsg(), &callDone, rpcClient);
if (ret != RPC_OK) {
pg_atomic_sub_fetch_u32(&ndpScan->reqCount, 1); // before send
mgr->ioFailed++;
return NdpRetCode::RPC_IO_SEND_FAILED;
}
#else
RpcStatus ret = RpcSendIOReq(req->GetReqMsg(), req->GetRespMsg(), NULL, rpcClient);
if (ret != RPC_OK) {
mgr->ioFailed++;
return NdpRetCode::RPC_IO_SEND_FAILED;
}
#endif
return NdpRetCode::NDP_OK;
}
NdpAdminRequest* NdpScanChannel::ConstructPlanReq(NdpScanDesc ndpScan)
{
if (IsA(ndpScan->cond->plan, Agg)) {
Agg* agg = (Agg*)ndpScan->cond->plan;
if (agg->grp_collations == NULL && agg->numCols > 0) {
agg->grp_collations = (unsigned int*)palloc(sizeof(unsigned int) * agg->numCols);
for (int i = 0; i < agg->numCols; i++) {
agg->grp_collations[i] = InvalidOid;
}
}
}
char* str = nodeToString(ndpScan->cond->plan);
int len = strlen(str) + 1;
if (len == 1) {
return nullptr;
}
NdpAdminRequest* req = reinterpret_cast<NdpAdminRequest*>(palloc(sizeof(NdpAdminRequest) + len));
req->head.command = NDP_PLAN;
req->head.size = sizeof(NdpAdminRequest) + len;
req->taskId = queryId;
req->tableId = ndpScan->cond->tableId;
errno_t rc = memcpy_s(reinterpret_cast<void*>(req + 1), len, str, len);
securec_check(rc, "", "");
return req;
}
bool NdpScanChannel::ExtractTupleDesc(TupleDesc desc, NdpTupleDesc* td)
{
td->natts = desc->natts;
if (td->natts == 0) {
td->attrs = NULL;
return true;
}
td->attrs = (NdpPGAttr *)palloc(sizeof(NdpPGAttr) * td->natts);
for (int i = 0; i < desc->natts; ++i) {
td->attrs[i].attlen = desc->attrs[i].attlen;
td->attrs[i].attbyval = desc->attrs[i].attbyval;
td->attrs[i].attcacheoff = desc->attrs[i].attcacheoff;
td->attrs[i].attalign = desc->attrs[i].attalign;
td->attrs[i].attndims = desc->attrs[i].attndims;
td->attrs[i].attstorage = desc->attrs[i].attstorage;
}
td->tdhasoid = desc->tdhasoid;
td->tdhasuids = desc->tdhasuids;
td->tdtypeid = desc->tdtypeid;
td->tdtypmod = desc->tdtypmod;
return true;
}
bool NdpScanChannel::ExtractRelation(TableScanDesc scan, NdpRelation* rel)
{
rel->node.spcNode = scan->rs_rd->rd_node.spcNode;
rel->node.dbNode = scan->rs_rd->rd_node.dbNode;
rel->node.relNode = scan->rs_rd->rd_node.relNode;
rel->node.bucketNode = scan->rs_rd->rd_node.bucketNode;
rel->node.opt = scan->rs_rd->rd_node.opt;
return ExtractTupleDesc(scan->rs_rd->rd_att, &rel->att);
}
bool NdpScanChannel::ExtractXact(TableScanDesc scan, NdpXact* xact)
{
xact->comboCids = NULL;
xact->CLogPageBuffer = NULL;
xact->CSNLogPageBuffer = NULL;
/* snapshot */
NdpSnapshot* snapshot = &xact->snapshot;
snapshot->satisfies = scan->rs_snapshot->satisfies;
snapshot->xmin = scan->rs_snapshot->xmin; /* all XID < xmin are visible to me */
snapshot->xmax = scan->rs_snapshot->xmax; /* all XID >= xmax are invisible to me */
snapshot->snapshotcsn = scan->rs_snapshot->snapshotcsn;
snapshot->curcid = scan->rs_snapshot->curcid;
/* TransactionState */
xact->transactionId = GetCurrentTransactionIdIfAny();
/* comboCids */
xact->usedComboCids = u_sess->utils_cxt.usedComboCids;
if (xact->usedComboCids > 0) {
xact->comboCids = (uint32 *)palloc(xact->usedComboCids * 2 * sizeof(uint32));
uint32 *comboCids = (uint32 *)u_sess->utils_cxt.comboCids;
for (int i = 0; i < xact->usedComboCids; i++) {
xact->comboCids[i * 2] = comboCids[i * 2];
xact->comboCids[i * 2 + 1] = comboCids[i * 2 + 1];
}
} else {
xact->comboCids = NULL;
}
/* clog & csnlog */
/* It's okay to read latestCompletedXid without acquiring ProcArrayLock shared lock
* because we dont' care if we get a slightly stale value
*/
xact->latestCompletedXid = t_thrd.xact_cxt.ShmemVariableCache->latestCompletedXid;
int64 pagenoStart, pagenoEnd, pageno;
pagenoStart = TransactionIdToPage(FirstNormalTransactionId);
pagenoEnd = TransactionIdToPage(xact->latestCompletedXid);
xact->CLogLen = (pagenoEnd - pagenoStart + 1) * BLCKSZ;
xact->CLogPageBuffer = (char *)palloc(xact->CLogLen);
for (pageno = pagenoStart; pageno <= pagenoEnd; pageno++) {
CopyCLog(pageno, xact->CLogPageBuffer + (pageno - pagenoStart) * BLCKSZ);
}
pagenoStart = TransactionIdToCSNPage(xact->latestCompletedXid);
pagenoEnd = TransactionIdToCSNPage(xact->latestCompletedXid);
xact->CSNLogLen = (pagenoEnd - pagenoStart + 1) * BLCKSZ;
xact->CSNLogPageBuffer = (char *)palloc(xact->CSNLogLen);
for (pageno = pagenoStart; pageno <= pagenoEnd; pageno++) {
CopyCSNLog(pageno, xact->CSNLogPageBuffer + (pageno - pagenoStart) * BLCKSZ);
}
return true;
}
bool NdpScanChannel::ExtractAggState(NdpScanDesc ndpScan, NdpAggState* aggS)
{
aggS->aggTd.natts = 0;
aggS->aggTd.attrs = nullptr;
aggS->aggNum = 0;
aggS->perAggTd = nullptr;
aggS->numCols = 0;
aggS->eqFuncOid = nullptr;
aggS->hashFuncOid = nullptr;
if (ndpScan->aggState) {
Assert(ndpScan->aggSlot != nullptr);
ExtractTupleDesc(ndpScan->aggSlot->tts_tupleDescriptor, &aggS->aggTd);
aggS->aggNum = ndpScan->aggState->numaggs;
aggS->perAggTd = (NdpTupleDesc*)palloc(aggS->aggNum * sizeof(NdpTupleDesc));
for (int aggNo = 0; aggNo < aggS->aggNum; ++aggNo) {
ExtractTupleDesc(ndpScan->aggState->evaldesc, &aggS->perAggTd[aggNo]);
}
Agg* agg = (Agg*)ndpScan->aggState->ss.ps.plan;
if ((agg->aggstrategy == AGG_HASHED) && (agg->numCols > 0)) {
aggS->numCols = agg->numCols;
aggS->eqFuncOid = (unsigned int*)palloc(aggS->numCols * sizeof(unsigned int));
aggS->hashFuncOid = (unsigned int*)palloc(aggS->numCols * sizeof(unsigned int));
for (int colNo = 0; colNo < aggS->numCols; ++colNo) {
aggS->eqFuncOid[colNo] = ndpScan->aggState->phases[0].eqfunctions[colNo].fn_oid;
aggS->hashFuncOid[colNo] = ndpScan->aggState->hashfunctions[colNo].fn_oid;
}
}
}
return true;
}
bool CheckExprContext(ExprContext* econtext, NdpParamList* pList)
{
ParamListInfo paramInfo = econtext->ecxt_param_list_info;
if (paramInfo->paramFetch != NULL) {
return false;
}
pList->numParams = paramInfo->numParams;
pList->params = (NdpParamData*)palloc(pList->numParams * sizeof(NdpParamData));
for (int i = 0; i < pList->numParams; i++) {
ParamExternData* from = &paramInfo->params[i];
NdpParamData* to = &pList->params[i];
if (from->tabInfo && from->tabInfo->isnestedtable && plpgsql_estate) {
pfree(pList->params);
pList->numParams = 0;
return false;
}
to->isnull = from->isnull;
to->ptype = from->ptype;
to->value = from->value;
get_typlenbyval(to->ptype, &to->typlen, &to->typbyval);
to->value = datumCopy(to->value, to->typbyval, to->typlen);
}
return true;
}
bool ExtractParamList(NdpScanDesc ndpScan, NdpParamList* pList)
{
ExprContext* econtext = ndpScan->scanState->ps.ps_ExprContext;
if (econtext == nullptr || econtext->ecxt_param_list_info == nullptr ||
econtext->ecxt_param_list_info->numParams == 0) {
pList->numParams = 0;
pList->params = nullptr;
return true;
}
return CheckExprContext(econtext, pList);
}
bool ExtractKnlSessionContext(NdpSessionContext* sess)
{
if (u_sess != nullptr) {
sess->sql_compatibility = u_sess->attr.attr_sql.sql_compatibility;
sess->behavior_compat_flags = u_sess->utils_cxt.behavior_compat_flags;
sess->encoding = u_sess->mb_cxt.DatabaseEncoding->encoding;
return true;
}
return false;
}
NdpPlanState* NdpScanChannel::CreatePlanState(NdpScanDesc ndpScan)
{
NdpPlanState* state = (NdpPlanState*)palloc(sizeof(NdpPlanState));
TableScanDesc scan = ndpScan->scan;
ExtractRelation(scan, &state->rel);
ProjectionInfo* proj_info = ndpScan->scanState->ps.ps_ProjInfo;
if (proj_info != NULL) {
ExtractTupleDesc(proj_info->pi_slot->tts_tupleDescriptor, &state->scanTd);
} else {
ExtractTupleDesc(ndpScan->scanState->ss_ScanTupleSlot->tts_tupleDescriptor, &state->scanTd);
}
if (!ExtractAggState(ndpScan, &state->aggState)) {
DestroyPlanState(state);
return nullptr;
}
if (!ExtractParamList(ndpScan, &state->paramList)) {
DestroyPlanState(state);
return nullptr;
}
if (!ExtractKnlSessionContext(&state->sess)) {
DestroyPlanState(state);
return nullptr;
}
return state;
}
void NdpScanChannel::DestroyPlanState(NdpPlanState* state)
{
if (state == nullptr) return;
if (state->rel.att.attrs) pfree(state->rel.att.attrs);
if (state->scanTd.attrs) pfree(state->scanTd.attrs);
if (state->paramList.params != nullptr) {
pfree(state->paramList.params);
}
NdpAggState* aggS = &state->aggState;
if (aggS->aggTd.attrs) pfree(aggS->aggTd.attrs);
if (aggS->perAggTd) {
for (int i = 0; i < aggS->aggNum; ++i) {
if (aggS->perAggTd[i].attrs) pfree(aggS->perAggTd[i].attrs);
}
pfree(aggS->perAggTd);
}
if (aggS->eqFuncOid) pfree(aggS->eqFuncOid);
if (aggS->hashFuncOid) pfree(aggS->hashFuncOid);
pfree(state);
}
NdpAdminRequest* NdpScanChannel::ConstructPlanState(NdpScanDesc ndpScan)
{
NdpPlanState* state = CreatePlanState(ndpScan);
if (state == nullptr) {
return nullptr;
}
StringInfoData str;
initStringInfo(&str);
NdpAdminRequest head;
appendBinaryStringInfo(&str, (const char*)&head, (int)sizeof(head));
stateToString(state, &str);
NdpAdminRequest* ptr = (NdpAdminRequest*)str.data;
ptr->head.command = NDP_PLANSTATE;
ptr->head.size = str.len;
ptr->taskId = queryId;
ptr->tableId = ndpScan->cond->tableId;
DestroyPlanState(state);
return ptr;
}
NdpAdminRequest* NdpScanChannel::ConstructQuery(NdpScanDesc ndpScan)
{
NdpQuery* query = (NdpQuery*)palloc(sizeof(NdpQuery));
NdpContext* context = static_cast<NdpContext*>(ndpScan->cond->ctx);
query->tableNum = context->tableCount;
ExtractXact(ndpScan->scan, &query->xact);
StringInfoData str;
initStringInfo(&str);
NdpAdminRequest head;
appendBinaryStringInfo(&str, (const char*)&head, (int)sizeof(head));
queryToString(query, &str);
NdpAdminRequest* ptr = (NdpAdminRequest*)str.data;
ptr->head.command = NDP_QUERY;
ptr->head.size = str.len;
if (query->xact.comboCids) pfree(query->xact.comboCids);
if (query->xact.CLogPageBuffer) pfree(query->xact.CLogPageBuffer);
if (query->xact.CSNLogPageBuffer) pfree(query->xact.CSNLogPageBuffer);
pfree(query);
return ptr;
}
NdpAdminRequest* NdpScanChannel::ConstructVersion()
{
int len = sizeof(uint64);
uint64 version = (((uint64)GRAND_VERSION_NUM) << 32) | NDP_LOCAL_VERSION_NUM;
NdpAdminRequest* req = reinterpret_cast<NdpAdminRequest*>(palloc(sizeof(NdpAdminRequest) + len));
req->head.command = NDP_VERSION;
req->head.size = sizeof(NdpAdminRequest) + len;
req->taskId = -1;
req->tableId = -1;
errno_t rc = memcpy_s(reinterpret_cast<void*>(req + 1), len, &version, len);
securec_check(rc, "", "");
return req;
}
Reference in New Issue View Git Blame Copy Permalink