This is an automated email from the ASF dual-hosted git repository.
tuhaihe pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cloudberry.git
The following commit(s) were added to refs/heads/main by this push:
new 08e3fc294fc Add views for getting cumulative IC stats (#1772)
08e3fc294fc is described below
commit 08e3fc294fcb0dcfe51732d60e127164ba8212fb
Author: Sagittarius <[email protected]>
AuthorDate: Mon Jun 22 12:52:18 2026 +0300
Add views for getting cumulative IC stats (#1772)
Main changes:
* `gp_interconnect_stats` — aggregated statistics across all segments;
* `gp_interconnect_stats_per_segment` — statistics grouped by segment;
* `gp_interconnect_stats_per_segment_per_host` — statistics grouped by host
and segment.
Based on the implementation from OpenGPDB
([open‑gpdb/gpdb#109](https://github.com/open-gpdb/gpdb/pull/109)).
---
.github/workflows/build-cloudberry.yml | 1 +
contrib/interconnect/Makefile | 6 ++
contrib/interconnect/README.md | 32 +++++++
contrib/interconnect/expected/interconnect.out | 82 +++++++++++++++++
contrib/interconnect/ic_modules.c | 38 ++++++++
contrib/interconnect/ic_modules.h | 1 +
contrib/interconnect/interconnect--1.0.sql | 110 +++++++++++++++++++++++
contrib/interconnect/interconnect.control | 4 +
contrib/interconnect/sql/interconnect.sql | 87 ++++++++++++++++++
contrib/interconnect/udp/ic_udpifc.c | 120 +++++++++++++++++++++++++
contrib/interconnect/udp/ic_udpifc.h | 28 ++++++
11 files changed, 509 insertions(+)
diff --git a/.github/workflows/build-cloudberry.yml
b/.github/workflows/build-cloudberry.yml
index c00dcde0486..0d76fa0da51 100644
--- a/.github/workflows/build-cloudberry.yml
+++ b/.github/workflows/build-cloudberry.yml
@@ -302,6 +302,7 @@ jobs:
"contrib/formatter_fixedwidth:installcheck",
"contrib/hstore:installcheck",
"contrib/indexscan:installcheck",
+ "contrib/interconnect:installcheck",
"contrib/pg_trgm:installcheck",
"contrib/indexscan:installcheck",
"contrib/pgcrypto:installcheck",
diff --git a/contrib/interconnect/Makefile b/contrib/interconnect/Makefile
index 31489fa9148..5bb0cd188e4 100644
--- a/contrib/interconnect/Makefile
+++ b/contrib/interconnect/Makefile
@@ -7,6 +7,10 @@ include
$(top_builddir)/contrib/interconnect/Makefile.interconnect
MODULE_big = interconnect
PGFILEDESC = "interconnect - inter connection module"
+EXTENSION = interconnect
+EXTENSION_VERSION = 1.0
+DATA = interconnect--$(EXTENSION_VERSION).sql
+
OBJS = \
$(WIN32RES) \
ic_common.o \
@@ -33,6 +37,8 @@ OBJS += proxy/ic_proxy_iobuf.o
SHLIB_LINK += $(filter -luv, $(LIBS))
endif # enable_ic_proxy
+REGRESS = interconnect
+
ifdef USE_PGXS
PG_CONFIG = pg_config
PGXS := $(shell $(PG_CONFIG) --pgxs)
diff --git a/contrib/interconnect/README.md b/contrib/interconnect/README.md
index fd9615d89c9..1ed7127969b 100644
--- a/contrib/interconnect/README.md
+++ b/contrib/interconnect/README.md
@@ -271,3 +271,35 @@ udpifc result:
Notice that: Lower TPS does not mean the protocol is slower, might means that
the cpu time taken by the protocol is low. For the udpifc, it satisfies the
highest tps required by `cbdb`. at the same time it occupies a lower cpu than
other types of interconnect.
+# interconnect statistics
+
+This extension provides cumulative interconnect statistics for Apache
Cloudberry, including queue sizes, buffer usage, retransmits, packet errors,
and other UDPIFC‑related metrics.
+
+It exposes three views with statistics at different aggregation levels:
+- gp_interconnect_stats — total cluster‑wide stats;
+- gp_interconnect_stats_per_segment — stats per segment;
+- gp_interconnect_stats_per_host — stats grouped by host.
+
+## How to create the extension
+
+Add interconnect to shared_preload_libraries and restart the cluster.
+
+```
+gpconfig -c shared_preload_libraries -v \
+ "$(psql -At -c \
+ "SELECT array_to_string( \
+ array_append( \
+ string_to_array( \
+ current_setting('shared_preload_libraries'), \
+ ','), \
+ 'interconnect'), \
+ ',')" \
+ postgres)"
+gpstop -ra
+```
+
+Create the extension in your database.
+
+```
+CREATE EXTENSION interconnect;
+```
diff --git a/contrib/interconnect/expected/interconnect.out
b/contrib/interconnect/expected/interconnect.out
new file mode 100644
index 00000000000..924a532e180
--- /dev/null
+++ b/contrib/interconnect/expected/interconnect.out
@@ -0,0 +1,82 @@
+-- Capture current interconnect stats as baseline for future comparisons
+SELECT * FROM gp_interconnect_stats \gset prev_
+-- Verify that all baseline interconnect statistics are >= 0 (no negative
values)
+SELECT
+ :prev_total_recv_queue_size >= 0,
+ :prev_recv_queue_conting_time >= 0,
+ :prev_total_capacity >= 0,
+ :prev_capacity_counting_time >= 0,
+ :prev_total_buffers >= 0,
+ :prev_buffer_counting_time >= 0,
+ :prev_retransmits >= 0,
+ :prev_startup_cached_pkts >= 0,
+ :prev_mismatches >= 0,
+ :prev_crs_errors >= 0,
+ :prev_snd_pkt_num >= 0,
+ :prev_recv_pkt_num >= 0,
+ :prev_disordered_pkt_num >= 0,
+ :prev_duplicate_pkt_num >= 0,
+ :prev_recv_ack_num >= 0,
+ :prev_status_query_msg_num >= 0;
+ ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? |
?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? |
?column? | ?column?
+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------
+ t | t | t | t | t | t | t |
t | t | t | t | t | t | t | t
| t
+(1 row)
+
+-- Create test table to generate interconnect traffic
+CREATE TABLE test_ic_data
+AS SELECT generate_series(1, 1000) AS id
+DISTRIBUTED RANDOMLY;
+-- Re-capture current state: overwrite prev with latest values
+SELECT * FROM gp_interconnect_stats \gset prev2_
+-- Check if current statistics are >= baseline values after first data
insertion
+SELECT
+ :prev2_total_recv_queue_size >= :prev_total_recv_queue_size,
+ :prev2_recv_queue_conting_time >= :prev_recv_queue_conting_time,
+ :prev2_total_capacity >= :prev_total_capacity,
+ :prev2_capacity_counting_time >= :prev_capacity_counting_time,
+ :prev2_total_buffers >= :prev_total_buffers,
+ :prev2_buffer_counting_time >= :prev_buffer_counting_time,
+ :prev2_retransmits >= :prev_retransmits,
+ :prev2_startup_cached_pkts >= :prev_startup_cached_pkts,
+ :prev2_mismatches >= :prev_mismatches,
+ :prev2_crs_errors >= :prev_crs_errors,
+ :prev2_snd_pkt_num >= :prev_snd_pkt_num,
+ :prev2_recv_pkt_num >= :prev_recv_pkt_num,
+ :prev2_disordered_pkt_num >= :prev_disordered_pkt_num,
+ :prev2_duplicate_pkt_num >= :prev_duplicate_pkt_num,
+ :prev2_recv_ack_num >= :prev_recv_ack_num,
+ :prev2_status_query_msg_num >= :prev_status_query_msg_num;
+ ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? |
?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? |
?column? | ?column?
+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------
+ t | t | t | t | t | t | t |
t | t | t | t | t | t | t | t
| t
+(1 row)
+
+-- Insert additional data to further test interconnect statistics changes
under load
+INSERT INTO test_ic_data SELECT generate_series(1001, 2000);
+-- Re‑check if current statistics remain >= baseline after second data
insertion
+SELECT
+ total_recv_queue_size >= :prev2_total_recv_queue_size,
+ recv_queue_conting_time >= :prev2_recv_queue_conting_time,
+ total_capacity >= :prev2_total_capacity,
+ capacity_counting_time >= :prev2_capacity_counting_time,
+ total_buffers >= :prev2_total_buffers,
+ buffer_counting_time >= :prev2_buffer_counting_time,
+ retransmits >= :prev2_retransmits,
+ startup_cached_pkts >= :prev2_startup_cached_pkts,
+ mismatches >= :prev2_mismatches,
+ crs_errors >= :prev2_crs_errors,
+ snd_pkt_num >= :prev2_snd_pkt_num,
+ recv_pkt_num >= :prev2_recv_pkt_num,
+ disordered_pkt_num >= :prev2_disordered_pkt_num,
+ duplicate_pkt_num >= :prev2_duplicate_pkt_num,
+ recv_ack_num >= :prev2_recv_ack_num,
+ status_query_msg_num >= :prev2_status_query_msg_num
+FROM gp_interconnect_stats;
+ ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? |
?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? |
?column? | ?column?
+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------
+ t | t | t | t | t | t | t |
t | t | t | t | t | t | t | t
| t
+(1 row)
+
+DROP TABLE test_ic_data;
+DROP EXTENSION interconnect;
diff --git a/contrib/interconnect/ic_modules.c
b/contrib/interconnect/ic_modules.c
index b582e8bdbe0..ef82777de12 100644
--- a/contrib/interconnect/ic_modules.c
+++ b/contrib/interconnect/ic_modules.c
@@ -16,6 +16,7 @@
#include "ic_common.h"
#include "tcp/ic_tcp.h"
#include "udp/ic_udpifc.h"
+#include "storage/ipc.h"
#ifdef ENABLE_IC_PROXY
#include "proxy/ic_proxy_server.h"
@@ -23,6 +24,9 @@
PG_MODULE_MAGIC;
+static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
+static shmem_request_hook_type prev_shmem_request_hook = NULL;
+
MotionIPCLayer tcp_ipc_layer = {
.ic_type = INTERCONNECT_TYPE_TCP,
.type_name = "tcp",
@@ -141,6 +145,24 @@ MotionIPCLayer udpifc_ipc_layer = {
.GetMotionSentRecordTypmod = GetMotionSentRecordTypmod,
};
+static void
+InterconnectShmemRequest(void)
+{
+ if (prev_shmem_request_hook)
+ prev_shmem_request_hook();
+
+ RequestAddinShmemSpace(MAXALIGN(sizeof(ICStatisticsShmem)));
+}
+
+static void
+InterconnectShmemInit(void)
+{
+ if (prev_shmem_startup_hook)
+ prev_shmem_startup_hook();
+
+ InterconnectShmemInitUDPIFC();
+}
+
void
_PG_init(void)
{
@@ -153,4 +175,20 @@ _PG_init(void)
RegisterIPCLayerImpl(&tcp_ipc_layer);
RegisterIPCLayerImpl(&udpifc_ipc_layer);
RegisterIPCLayerImpl(&proxy_ipc_layer);
+
+ if (Gp_interconnect_type == INTERCONNECT_TYPE_UDPIFC)
+ {
+ prev_shmem_request_hook = shmem_request_hook;
+ shmem_request_hook = InterconnectShmemRequest;
+
+ prev_shmem_startup_hook = shmem_startup_hook;
+ shmem_startup_hook = InterconnectShmemInit;
+ }
+}
+
+void
+_PG_fini(void)
+{
+ shmem_request_hook = prev_shmem_request_hook;
+ shmem_startup_hook = prev_shmem_startup_hook;
}
diff --git a/contrib/interconnect/ic_modules.h
b/contrib/interconnect/ic_modules.h
index a381d279fdd..e73b1998f3a 100644
--- a/contrib/interconnect/ic_modules.h
+++ b/contrib/interconnect/ic_modules.h
@@ -18,5 +18,6 @@ extern MotionIPCLayer proxy_ipc_layer;
extern MotionIPCLayer udpifc_ipc_layer;
extern void _PG_init(void);
+extern void _PG_fini(void);
#endif // INTER_CONNECT_H
diff --git a/contrib/interconnect/interconnect--1.0.sql
b/contrib/interconnect/interconnect--1.0.sql
new file mode 100644
index 00000000000..f4aa635583e
--- /dev/null
+++ b/contrib/interconnect/interconnect--1.0.sql
@@ -0,0 +1,110 @@
+/* contrib/interconnect/interconnect--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION interconnect" to load this file. \quit
+
+CREATE FUNCTION __gp_interconnect_get_stats_f_on_master(
+ OUT gp_segment_id smallint,
+ OUT total_recv_queue_size bigint,
+ OUT recv_queue_conting_time bigint,
+ OUT total_capacity bigint,
+ OUT capacity_counting_time bigint,
+ OUT total_buffers bigint,
+ OUT buffer_counting_time bigint,
+ OUT retransmits bigint,
+ OUT startup_cached_pkts bigint,
+ OUT mismatches bigint,
+ OUT crs_errors bigint,
+ OUT snd_pkt_num bigint,
+ OUT recv_pkt_num bigint,
+ OUT disordered_pkt_num bigint,
+ OUT duplicate_pkt_num bigint,
+ OUT recv_ack_num bigint,
+ OUT status_query_msg_num bigint
+)
+RETURNS SETOF record
+LANGUAGE C VOLATILE EXECUTE ON MASTER
+AS '$libdir/interconnect', 'gp_interconnect_get_stats';
+
+CREATE FUNCTION __gp_interconnect_get_stats_f_on_segments(
+ OUT gp_segment_id smallint,
+ OUT total_recv_queue_size bigint,
+ OUT recv_queue_conting_time bigint,
+ OUT total_capacity bigint,
+ OUT capacity_counting_time bigint,
+ OUT total_buffers bigint,
+ OUT buffer_counting_time bigint,
+ OUT retransmits bigint,
+ OUT startup_cached_pkts bigint,
+ OUT mismatches bigint,
+ OUT crs_errors bigint,
+ OUT snd_pkt_num bigint,
+ OUT recv_pkt_num bigint,
+ OUT disordered_pkt_num bigint,
+ OUT duplicate_pkt_num bigint,
+ OUT recv_ack_num bigint,
+ OUT status_query_msg_num bigint
+)
+RETURNS SETOF record LANGUAGE C VOLATILE EXECUTE ON ALL SEGMENTS
+AS '$libdir/interconnect', 'gp_interconnect_get_stats';
+
+
+-- Cummulative interconnect statistics per segment
+CREATE VIEW gp_interconnect_stats_per_segment AS
+ SELECT c.hostname, s.* FROM (
+ SELECT * FROM __gp_interconnect_get_stats_f_on_master()
+ UNION ALL
+ SELECT * FROM __gp_interconnect_get_stats_f_on_segments()
+ ) s
+ JOIN pg_catalog.gp_segment_configuration AS c
+ ON s.gp_segment_id = c.content AND c.role = 'p';
+
+GRANT SELECT ON gp_interconnect_stats_per_segment TO public;
+
+-- Cummulative interconnect statistics
+CREATE VIEW gp_interconnect_stats AS
+ SELECT
+ sum(total_recv_queue_size) as total_recv_queue_size
+ , sum(recv_queue_conting_time) as recv_queue_conting_time
+ , sum(total_capacity) as total_capacity
+ , sum(capacity_counting_time) as capacity_counting_time
+ , sum(total_buffers) as total_buffers
+ , sum(buffer_counting_time) as buffer_counting_time
+ , sum(retransmits) as retransmits
+ , sum(startup_cached_pkts) as startup_cached_pkts
+ , sum(mismatches) as mismatches
+ , sum(crs_errors) as crs_errors
+ , sum(snd_pkt_num) as snd_pkt_num
+ , sum(recv_pkt_num) as recv_pkt_num
+ , sum(disordered_pkt_num) as disordered_pkt_num
+ , sum(duplicate_pkt_num) as duplicate_pkt_num
+ , sum(recv_ack_num) as recv_ack_num
+ , sum(status_query_msg_num) as status_query_msg_num
+ FROM gp_interconnect_stats_per_segment;
+
+GRANT SELECT ON gp_interconnect_stats TO public;
+
+-- Cummulative interconnect statistics grouped by host
+CREATE VIEW gp_interconnect_stats_per_host AS
+ SELECT
+ hostname
+ , sum(total_recv_queue_size) as total_recv_queue_size
+ , sum(recv_queue_conting_time) as recv_queue_conting_time
+ , sum(total_capacity) as total_capacity
+ , sum(capacity_counting_time) as capacity_counting_time
+ , sum(total_buffers) as total_buffers
+ , sum(buffer_counting_time) as buffer_counting_time
+ , sum(retransmits) as retransmits
+ , sum(startup_cached_pkts) as startup_cached_pkts
+ , sum(mismatches) as mismatches
+ , sum(crs_errors) as crs_errors
+ , sum(snd_pkt_num) as snd_pkt_num
+ , sum(recv_pkt_num) as recv_pkt_num
+ , sum(disordered_pkt_num) as disordered_pkt_num
+ , sum(duplicate_pkt_num) as duplicate_pkt_num
+ , sum(recv_ack_num) as recv_ack_num
+ , sum(status_query_msg_num) as status_query_msg_num
+ FROM gp_interconnect_stats_per_segment
+ GROUP BY hostname;
+
+GRANT SELECT ON gp_interconnect_stats_per_host TO public;
diff --git a/contrib/interconnect/interconnect.control
b/contrib/interconnect/interconnect.control
new file mode 100644
index 00000000000..0cf63e411c9
--- /dev/null
+++ b/contrib/interconnect/interconnect.control
@@ -0,0 +1,4 @@
+comment = 'Cummulative statistics from UDPIFC interconnect protocol'
+default_version = '1.0'
+relocatable = false
+schema = public
diff --git a/contrib/interconnect/sql/interconnect.sql
b/contrib/interconnect/sql/interconnect.sql
new file mode 100644
index 00000000000..4e6555b6b82
--- /dev/null
+++ b/contrib/interconnect/sql/interconnect.sql
@@ -0,0 +1,87 @@
+-- start_ignore
+\! gpconfig -c shared_preload_libraries -v "interconnect"
+\! gpstop -raiq
+\c
+DROP TABLE IF EXISTS test_ic_data;
+CREATE EXTENSION IF NOT EXISTS interconnect;
+-- end_ignore
+
+-- Capture current interconnect stats as baseline for future comparisons
+SELECT * FROM gp_interconnect_stats \gset prev_
+
+-- Verify that all baseline interconnect statistics are >= 0 (no negative
values)
+SELECT
+ :prev_total_recv_queue_size >= 0,
+ :prev_recv_queue_conting_time >= 0,
+ :prev_total_capacity >= 0,
+ :prev_capacity_counting_time >= 0,
+ :prev_total_buffers >= 0,
+ :prev_buffer_counting_time >= 0,
+ :prev_retransmits >= 0,
+ :prev_startup_cached_pkts >= 0,
+ :prev_mismatches >= 0,
+ :prev_crs_errors >= 0,
+ :prev_snd_pkt_num >= 0,
+ :prev_recv_pkt_num >= 0,
+ :prev_disordered_pkt_num >= 0,
+ :prev_duplicate_pkt_num >= 0,
+ :prev_recv_ack_num >= 0,
+ :prev_status_query_msg_num >= 0;
+
+-- Create test table to generate interconnect traffic
+CREATE TABLE test_ic_data
+AS SELECT generate_series(1, 1000) AS id
+DISTRIBUTED RANDOMLY;
+
+-- Re-capture current state: overwrite prev with latest values
+SELECT * FROM gp_interconnect_stats \gset prev2_
+
+-- Check if current statistics are >= baseline values after first data
insertion
+SELECT
+ :prev2_total_recv_queue_size >= :prev_total_recv_queue_size,
+ :prev2_recv_queue_conting_time >= :prev_recv_queue_conting_time,
+ :prev2_total_capacity >= :prev_total_capacity,
+ :prev2_capacity_counting_time >= :prev_capacity_counting_time,
+ :prev2_total_buffers >= :prev_total_buffers,
+ :prev2_buffer_counting_time >= :prev_buffer_counting_time,
+ :prev2_retransmits >= :prev_retransmits,
+ :prev2_startup_cached_pkts >= :prev_startup_cached_pkts,
+ :prev2_mismatches >= :prev_mismatches,
+ :prev2_crs_errors >= :prev_crs_errors,
+ :prev2_snd_pkt_num >= :prev_snd_pkt_num,
+ :prev2_recv_pkt_num >= :prev_recv_pkt_num,
+ :prev2_disordered_pkt_num >= :prev_disordered_pkt_num,
+ :prev2_duplicate_pkt_num >= :prev_duplicate_pkt_num,
+ :prev2_recv_ack_num >= :prev_recv_ack_num,
+ :prev2_status_query_msg_num >= :prev_status_query_msg_num;
+
+-- Insert additional data to further test interconnect statistics changes
under load
+INSERT INTO test_ic_data SELECT generate_series(1001, 2000);
+
+-- Re‑check if current statistics remain >= baseline after second data
insertion
+SELECT
+ total_recv_queue_size >= :prev2_total_recv_queue_size,
+ recv_queue_conting_time >= :prev2_recv_queue_conting_time,
+ total_capacity >= :prev2_total_capacity,
+ capacity_counting_time >= :prev2_capacity_counting_time,
+ total_buffers >= :prev2_total_buffers,
+ buffer_counting_time >= :prev2_buffer_counting_time,
+ retransmits >= :prev2_retransmits,
+ startup_cached_pkts >= :prev2_startup_cached_pkts,
+ mismatches >= :prev2_mismatches,
+ crs_errors >= :prev2_crs_errors,
+ snd_pkt_num >= :prev2_snd_pkt_num,
+ recv_pkt_num >= :prev2_recv_pkt_num,
+ disordered_pkt_num >= :prev2_disordered_pkt_num,
+ duplicate_pkt_num >= :prev2_duplicate_pkt_num,
+ recv_ack_num >= :prev2_recv_ack_num,
+ status_query_msg_num >= :prev2_status_query_msg_num
+FROM gp_interconnect_stats;
+
+DROP TABLE test_ic_data;
+DROP EXTENSION interconnect;
+
+-- start_ignore
+\! gpconfig -r shared_preload_libraries
+\! gpstop -raiq
+-- end_ignore
diff --git a/contrib/interconnect/udp/ic_udpifc.c
b/contrib/interconnect/udp/ic_udpifc.c
index be14218f168..47c6273470c 100644
--- a/contrib/interconnect/udp/ic_udpifc.c
+++ b/contrib/interconnect/udp/ic_udpifc.c
@@ -41,6 +41,7 @@
#include "access/transam.h"
#include "access/xact.h"
#include "common/ip.h"
+#include "funcapi.h"
#include "nodes/execnodes.h"
#include "nodes/pg_list.h"
#include "nodes/print.h"
@@ -51,6 +52,8 @@
#include "pgstat.h"
#include "postmaster/postmaster.h"
#include "storage/latch.h"
+#include "storage/lock.h"
+#include "storage/pg_shmem.h"
#include "storage/pmsignal.h"
#include "utils/builtins.h"
#include "utils/guc.h"
@@ -721,6 +724,8 @@ typedef struct ICStatistics
/* Statistics for UDP interconnect. */
static ICStatistics ic_statistics;
+static ICStatisticsShmem *pICStatisticsShmem = NULL;
+
/* UDP listen fd */
int UDP_listenerFd;
@@ -1821,6 +1826,27 @@ ic_reset_pthread_sigmasks(sigset_t *sigs)
return;
}
+void
+InterconnectShmemInitUDPIFC(void)
+{
+ bool found;
+ pICStatisticsShmem = ShmemInitStruct("global interconnect statistics",
+ sizeof(ICStatisticsShmem), &found);
+ if (pICStatisticsShmem == NULL)
+ {
+ ereport(FATAL,
+ (errcode(ERRCODE_OUT_OF_MEMORY),
+ errmsg("not enough shared memory for global
interconnect statistics")));
+ }
+
+ if (!found)
+ memset(pICStatisticsShmem, 0, sizeof(*pICStatisticsShmem));
+
+ int tranche_id = LWLockNewTrancheId();
+ LWLockRegisterTranche(tranche_id, "IC Statistics");
+ LWLockInitialize(&pICStatisticsShmem->lock, tranche_id);
+}
+
/*
* InitMotionUDPIFC
* Initialize UDP specific comms, and create rx-thread.
@@ -3908,6 +3934,30 @@ chunkTransportStateEntryInitialized(ChunkTransportState
*transportStates,
return pEntry->valid;
}
+/* Append local interconnect stats to global cummulative stats. */
+static void
+updateGlobalInterconnectStats(void)
+{
+ LWLockAcquire(&pICStatisticsShmem->lock, LW_EXCLUSIVE);
+ pICStatisticsShmem->totalRecvQueueSize +=
ic_statistics.totalRecvQueueSize;
+ pICStatisticsShmem->recvQueueSizeCountingTime +=
ic_statistics.recvQueueSizeCountingTime;
+ pICStatisticsShmem->totalCapacity += ic_statistics.totalCapacity;
+ pICStatisticsShmem->capacityCountingTime +=
ic_statistics.capacityCountingTime;
+ pICStatisticsShmem->totalBuffers += ic_statistics.totalBuffers;
+ pICStatisticsShmem->bufferCountingTime +=
ic_statistics.bufferCountingTime;
+ pICStatisticsShmem->retransmits += ic_statistics.retransmits;
+ pICStatisticsShmem->startupCachedPktNum +=
ic_statistics.startupCachedPktNum;
+ pICStatisticsShmem->mismatchNum += ic_statistics.mismatchNum;
+ pICStatisticsShmem->crcErrors += ic_statistics.crcErrors;
+ pICStatisticsShmem->sndPktNum += ic_statistics.sndPktNum;
+ pICStatisticsShmem->recvPktNum += ic_statistics.recvPktNum;
+ pICStatisticsShmem->disorderedPktNum += ic_statistics.disorderedPktNum;
+ pICStatisticsShmem->duplicatedPktNum += ic_statistics.duplicatedPktNum;
+ pICStatisticsShmem->recvAckNum += ic_statistics.recvAckNum;
+ pICStatisticsShmem->statusQueryMsgNum +=
ic_statistics.statusQueryMsgNum;
+ LWLockRelease(&pICStatisticsShmem->lock);
+}
+
/*
* computeNetworkStatistics
* Compute the max/min/avg network statistics.
@@ -4214,6 +4264,7 @@ TeardownUDPIFCInterconnect_Internal(ChunkTransportState
*transportStates,
(minRtt == ~((uint64) 0) ? 0 : minRtt), (minDev == ~((uint64)
0) ? 0 : minDev), avgRtt, avgDev, maxRtt, maxDev,
snd_control_info.cwnd, ic_statistics.statusQueryMsgNum);
+ updateGlobalInterconnectStats();
ic_control_info.isSender = false;
memset(&ic_statistics, 0, sizeof(ICStatistics));
@@ -8231,3 +8282,72 @@ MlPutRxBufferIFC(ChunkTransportState *transportStates,
int motNodeID, int route)
if (param.msg.len != 0)
sendAckWithParam(¶m);
}
+
+PG_FUNCTION_INFO_V1(gp_interconnect_get_stats);
+
+Datum
+gp_interconnect_get_stats(PG_FUNCTION_ARGS)
+{
+ if (Gp_interconnect_type != INTERCONNECT_TYPE_UDPIFC)
+ {
+ ereport(WARNING,
+ (errcode(ERRCODE_WARNING_GP_INTERCONNECTION),
+ errmsg("Interconnect statistics are collected only for UDPIFC
protocol")));
+ PG_RETURN_NULL();
+ }
+
+ /*
+ * Build a tuple descriptor for our result type
+ * The number and type of attributes have to match the definition of the
+ * view gp_interconnect_stats_per_segment
+ */
+ enum {NUM_IC_STATS_ELEM = 17};
+ TupleDesc tupdesc = CreateTemplateTupleDesc(NUM_IC_STATS_ELEM);
+
+ TupleDescInitEntry(tupdesc, 1, "segid", INT2OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 2, "total_recv_queue_size", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 3, "recv_queue_conting_time", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 4, "total_capacity", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 5, "capacity_counting_time", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 6, "total_buffers", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 7, "buffer_counting_time", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 8, "retransmits", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 9, "startup_cached_pkts", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 10, "mismatches", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 11, "crs_errors", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 12, "snd_pkt_num", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 13, "recv_pkt_num", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 14, "disordered_pkt_num", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 15, "duplicate_pkt_num", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 16, "recv_ack_num", INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, 17, "status_query_msg_num", INT8OID, -1, 0);
+ tupdesc = BlessTupleDesc(tupdesc);
+
+ Datum values[NUM_IC_STATS_ELEM];
+ bool nulls[NUM_IC_STATS_ELEM] = {0};
+
+ LWLockAcquire(&pICStatisticsShmem->lock, LW_SHARED);
+ values[0] = Int32GetDatum(GpIdentity.segindex);
+ values[1] = Int64GetDatum(pICStatisticsShmem->totalRecvQueueSize);
+ values[2] =
Int64GetDatum(pICStatisticsShmem->recvQueueSizeCountingTime);
+ values[3] = Int64GetDatum(pICStatisticsShmem->totalCapacity);
+ values[4] = Int64GetDatum(pICStatisticsShmem->capacityCountingTime);
+ values[5] = Int64GetDatum(pICStatisticsShmem->totalBuffers);
+ values[6] = Int64GetDatum(pICStatisticsShmem->bufferCountingTime);
+ values[7] = Int64GetDatum(pICStatisticsShmem->retransmits);
+ values[8] = Int64GetDatum(pICStatisticsShmem->startupCachedPktNum);
+ values[9] = Int64GetDatum(pICStatisticsShmem->mismatchNum);
+ values[10] = Int64GetDatum(pICStatisticsShmem->crcErrors);
+ values[11] = Int64GetDatum(pICStatisticsShmem->sndPktNum);
+ values[12] = Int64GetDatum(pICStatisticsShmem->recvPktNum);
+ values[13] = Int64GetDatum(pICStatisticsShmem->disorderedPktNum);
+ values[14] = Int64GetDatum(pICStatisticsShmem->duplicatedPktNum);
+ values[15] = Int64GetDatum(pICStatisticsShmem->recvAckNum);
+ values[16] = Int64GetDatum(pICStatisticsShmem->statusQueryMsgNum);
+ LWLockRelease(&pICStatisticsShmem->lock);
+
+ HeapTuple tuple = heap_form_tuple(tupdesc, values, nulls);
+ Datum result = HeapTupleGetDatum(tuple);
+
+ PG_RETURN_DATUM(result);
+}
diff --git a/contrib/interconnect/udp/ic_udpifc.h
b/contrib/interconnect/udp/ic_udpifc.h
index af3ca72ba3b..346f6ea85cb 100644
--- a/contrib/interconnect/udp/ic_udpifc.h
+++ b/contrib/interconnect/udp/ic_udpifc.h
@@ -17,6 +17,7 @@
#include "nodes/execnodes.h" /* ExecSlice, SliceTable */
#include "miscadmin.h"
#include "libpq/libpq-be.h"
+#include "storage/lwlock.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
@@ -212,4 +213,31 @@ extern void dumpICBufferList(ICBufferList * list, const
char *fname);
extern void dumpUnackQueueRing(const char *fname);
extern void dumpConnections(ChunkTransportStateEntry * pEntry, const char
*fname);
+/*
+ * Keeps various statistics about interconnect internal.
+ * Also those numbers are expected to grow big, hence uint64.
+ */
+typedef struct ICStatisticsShmem
+{
+ LWLock lock; /*
mutex for synchronizing access to statistics data */
+ uint64 totalRecvQueueSize; /* receive
queue size sum when main thread is trying to get a packet */
+ uint64 recvQueueSizeCountingTime; /* counting times when
computing totalRecvQueueSize */
+ uint64 totalCapacity; /* the capacity
sum when packets are tried to be sent */
+ uint64 capacityCountingTime; /* counting times used
to compute totalCapacity */
+ uint64 totalBuffers; /* total
buffers available when sending packets */
+ uint64 bufferCountingTime; /* counting
times when compute totalBuffers */
+ uint64 retransmits; /* the number
of packet retransmits */
+ uint64 startupCachedPktNum; /* number of packets
cached during connection startup */
+ uint64 mismatchNum; /* the number
of mismatched packets received */
+ uint64 crcErrors; /* the
number of crc errors */
+ uint64 sndPktNum; /* the
number of packets sent by sender */
+ uint64 recvPktNum; /* the
number of packets received by receiver */
+ uint64 disorderedPktNum; /* disordered
packet number */
+ uint64 duplicatedPktNum; /* duplicate
packet number */
+ uint64 recvAckNum; /* the
number of Acks received */
+ uint64 statusQueryMsgNum; /* the number
of status query messages sent */
+} ICStatisticsShmem;
+
+void InterconnectShmemInitUDPIFC(void);
+
#endif // IC_UDP_INTERFACE_H
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
