add pg_stat_he3walwrite view

Signed-off-by: shipixian <shipixian_yewu@cmss.chinamobile.com>

src/backend/access/transam/xlog.c

@@ -673,8 +673,8 @@ typedef struct FlushInfo
 
 typedef struct XLogParralFlush
 {
-	pg_atomic_uint32 		begin;
-	uint32 		last;
+	pg_atomic_uint64 		begin;
+	uint64 		last;
 //	uint32		count;
 //	uint32		diff;
 	// FlushInfo 	wrtResult[128];
@@ -3459,7 +3459,7 @@ FlushWal(XLogwrtRqst WriteRqst)
 	uint64 count;
 	// int stp;
 	int xlogLength;
-	uint32 curLoc = 0;
+	uint64 curLoc = 0;
 	bool mustDo = false;
 	bool nowrite = false;
 	
@@ -3604,7 +3604,7 @@ mustflush:
 			// SpinLockRelease(&XLogCtl->info_lck);
 
 			// printf("end flush wals, begin %d, curLoc %d, WriteRqst.Write %ld\n", flushInfo.begin, curLoc, WriteRqst.Write);
-			while (pg_atomic_read_u32(&XLogCtl->LogFlush.begin) < curLoc)
+			while (pg_atomic_read_u64(&XLogCtl->LogFlush.begin) < curLoc)
 			{
 				pg_usleep(20L);
 
@@ -3636,7 +3636,7 @@ mustflush:
 			} else
 				XLogCtl->LogFlush.diff = diff;
 			*/
-			pg_atomic_write_u32(&XLogCtl->LogFlush.begin, curLoc+1);
+			pg_atomic_write_u64(&XLogCtl->LogFlush.begin, curLoc+1);
 			// SpinLockAcquire(&XLogCtl->info_lck);
 
 			// XLogCtl->LogFlush.begin = curLoc+1;
@@ -15777,3 +15777,16 @@ static void PrecacheHotDataByRules()
     PQclear(ruleRes);
     PQfinish(metaConn);
 }
+
+void He3DBGetWalWriteStats(XLogRecPtr *writtenlsn, XLogRecPtr *flushlsn, uint64 *totaltimes, int *parallels)
+{
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	LogwrtResult = XLogCtl->LogwrtResult;
+	XLogParralFlush flushinfo = XLogCtl->LogFlush;
+	SpinLockRelease(&XLogCtl->info_lck);
+	*writtenlsn = LogwrtResult.Write;
+	*flushlsn = (XLogRecPtr) pg_atomic_read_u64(&LogwrtResult.Flush);
+	*totaltimes = flushinfo.last;
+	*parallels = flushinfo.last - pg_atomic_read_u64(&flushinfo.begin);
+}

src/backend/catalog/system_views.sql

@@ -866,6 +866,15 @@ CREATE VIEW pg_stat_replication AS
         JOIN pg_stat_get_wal_senders() AS W ON (S.pid = W.pid)
         LEFT JOIN pg_authid AS U ON (S.usesysid = U.oid);
 
+CREATE VIEW pg_stat_he3walwrite AS
+    SELECT
+            s.write_lsn,
+            s.flush_lsn,
+            s.writekv_totaltimes,
+            s.writekv_parallels
+    FROM pg_stat_get_he3walwrite() AS s
+        ;
+
 CREATE VIEW pg_stat_slru AS
     SELECT
             s.name,

src/backend/utils/adt/pgstatfuncs.c

@@ -25,11 +25,13 @@
 #include "postmaster/bgworker_internals.h"
 #include "postmaster/postmaster.h"
 #include "replication/slot.h"
+#include "storage/bufmgr.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/inet.h"
+#include "utils/pg_lsn.h"
 #include "utils/timestamp.h"
 
 #define UINT32_ACCESS_ONCE(var)		 ((uint32)(*((volatile uint32 *)&(var))))
@@ -2381,3 +2383,50 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 	/* Returns the record as Datum */
 	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
 }
+
+/*
+ * Returns statistics of WAL activity
+ */
+Datum
+pg_stat_get_he3walwrite(PG_FUNCTION_ARGS)
+{
+#define PG_STAT_GET_HE3WALWRITE_COLS	4
+	TupleDesc	tupdesc;
+	Datum		values[PG_STAT_GET_HE3WALWRITE_COLS];
+	bool		nulls[PG_STAT_GET_HE3WALWRITE_COLS];
+	XLogRecPtr 	writtenlsn, flushlsn;
+	uint64		writtenTimes;
+	int			parallels;
+
+	/* Initialise values and NULL flags arrays */
+	MemSet(values, 0, sizeof(values));
+	MemSet(nulls, 0, sizeof(nulls));
+
+	/* Initialise attributes information in the tuple descriptor */
+	tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_HE3WALWRITE_COLS);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "write_lsn",
+					   PG_LSNOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "flush_lsn",
+					   PG_LSNOID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "writekv_totaltimes",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "writekv_parallels",
+					   INT4OID, -1, 0);
+
+	BlessTupleDesc(tupdesc);
+
+	/* Get statistics about WAL Write */
+	if (EnableHotStandby && *isPromoteIsTriggered == false)
+		PG_RETURN_NULL();
+	
+	He3DBGetWalWriteStats(&writtenlsn, &flushlsn, &writtenTimes, &parallels);
+
+	/* Fill values and NULLs */
+	values[0] = LSNGetDatum(writtenlsn);
+	values[1] = LSNGetDatum(flushlsn);
+	values[2] = UInt64GetDatum(writtenTimes);
+	values[3] = Int32GetDatum(parallels);
+
+	/* Returns the record as Datum */
+	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
+}

src/include/access/xlog.h

@@ -429,6 +429,7 @@ extern void do_pg_abort_backup(int code, Datum arg);
 extern void register_persistent_abort_backup_handler(void);
 extern SessionBackupState get_backup_status(void);
 extern void pushXlogToTikv(char*data,int len);
+extern void He3DBGetWalWriteStats(XLogRecPtr *writtenlsn, XLogRecPtr *flushlsn, uint64 *totaltimes, int *parallels);
 
 /* File path names (all relative to $PGDATA) */
 #define RECOVERY_SIGNAL_FILE	"recovery.signal"

src/include/catalog/pg_proc.dat

@@ -5582,6 +5582,14 @@
   proargnames => '{wal_records,wal_fpi,wal_bytes,wal_buffers_full,wal_write,wal_sync,wal_write_time,wal_sync_time,stats_reset}',
   prosrc => 'pg_stat_get_wal' },
 
+{ oid => '6206', descr => 'statistics: information about He3DB WAL Write',
+  proname => 'pg_stat_get_he3walwrite', proisstrict => 'f', provolatile => 's',
+  proparallel => 'r', prorettype => 'record', proargtypes => '',
+  proallargtypes => '{pg_lsn,pg_lsn,int8,int4}',
+  proargmodes => '{o,o,o,o}',
+  proargnames => '{write_lsn,flush_lsn,writekv_totaltimes,writekv_parallels}',
+  prosrc => 'pg_stat_get_he3walwrite' },
+
 { oid => '2306', descr => 'statistics: information about SLRU caches',
   proname => 'pg_stat_get_slru', prorows => '100', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',