Make Remote provider collect wire statistics.

New info items to query wire stats counters.

Author: hvlad

src/include/firebird/impl/inf_pub.h

@@ -178,6 +178,17 @@ enum db_info_types
 
 	fb_info_parallel_workers = 149,
 
+	// Wire stats items, implemented by Remote provider only
+	fb_info_wire_out_packets = 150,
+	fb_info_wire_in_packets = 151,
+	fb_info_wire_out_bytes = 152,
+	fb_info_wire_in_bytes = 153,
+	fb_info_wire_snd_packets = 154,
+	fb_info_wire_rcv_packets = 155,
+	fb_info_wire_snd_bytes = 156,
+	fb_info_wire_rcv_bytes = 157,
+	fb_info_wire_roundtrips = 158,
+
 	isc_info_db_last_value   /* Leave this LAST! */
 };
 

src/include/gen/Firebird.pas

@@ -4496,6 +4496,15 @@ 	IProfilerStatsImpl = class(IProfilerStats)
 	fb_info_username = byte(147);
 	fb_info_sqlrole = byte(148);
 	fb_info_parallel_workers = byte(149);
+	fb_info_wire_out_packets = byte(150);
+	fb_info_wire_in_packets = byte(151);
+	fb_info_wire_out_bytes = byte(152);
+	fb_info_wire_in_bytes = byte(153);
+	fb_info_wire_snd_packets = byte(154);
+	fb_info_wire_rcv_packets = byte(155);
+	fb_info_wire_snd_bytes = byte(156);
+	fb_info_wire_rcv_bytes = byte(157);
+	fb_info_wire_roundtrips = byte(158);
 	fb_info_crypt_encrypted = $01;
 	fb_info_crypt_process = $02;
 	fb_feature_multi_statements = byte(1);

src/remote/client/interface.cpp

@@ -936,6 +936,8 @@ class Attachment final : public RefCntIface<IAttachmentImpl<Attachment, CheckSta
 	void internalDetach(CheckStatusWrapper* status);
 	void internalDropDatabase(CheckStatusWrapper* status);
 	SLONG getSingleInfo(CheckStatusWrapper* status, UCHAR infoItem);
+	bool getWireStatsInfo(unsigned int item_length, const unsigned char* items,
+						  unsigned int buffer_length, unsigned char* buffer) const;
 
 	Rdb* rdb;
 	const PathName dbPath;
@@ -1966,6 +1968,55 @@ IAttachment* Loopback::createDatabase(CheckStatusWrapper* status, const char* fi
 }
 
 
+bool Attachment::getWireStatsInfo(unsigned int item_length, const unsigned char* items,
+								  unsigned int buffer_length, unsigned char* buffer) const
+{
+	const rem_port* const port = rdb->rdb_port;
+
+	UCHAR* ptr = buffer;
+	const UCHAR* const end = buffer + buffer_length;
+
+	for (auto item = items; item < items + item_length; item++)
+	{
+		switch (*item)
+		{
+		case fb_info_wire_snd_packets:
+		case fb_info_wire_rcv_packets:
+		case fb_info_wire_out_packets:
+		case fb_info_wire_in_packets:
+		case fb_info_wire_snd_bytes:
+		case fb_info_wire_rcv_bytes:
+		case fb_info_wire_out_bytes:
+		case fb_info_wire_in_bytes:
+		case fb_info_wire_roundtrips:
+		{
+			const FB_UINT64 value = port->getStatItem(*item);
+
+			if (value <= MAX_SLONG)
+				ptr = fb_utils::putInfoItemInt(*item, (SLONG) value, ptr, end);
+			else
+				ptr = fb_utils::putInfoItemInt(*item, value, ptr, end);
+
+			fb_assert(ptr);
+			break;
+		}
+
+		case isc_info_end:
+			if (ptr < end)
+				*ptr++ = *item;
+
+			return true;
+
+		default:
+			// Let someone else handle unknown item
+			return false;
+		}
+	}
+
+	return true;
+}
+
+
 void Attachment::getInfo(CheckStatusWrapper* status,
 						 unsigned int item_length, const unsigned char* items,
 						 unsigned int buffer_length, unsigned char* buffer)
@@ -1994,6 +2045,9 @@ void Attachment::getInfo(CheckStatusWrapper* status,
 
 		RefMutexGuard portGuard(*port->port_sync, FB_FUNCTION);
 
+		if (getWireStatsInfo(item_length, items, buffer_length, buffer))
+			return;
+
 		UCHAR* temp_buffer = temp.getBuffer(buffer_length);
 
 		info(status, rdb, op_info_database, rdb->rdb_id, 0,

src/remote/inet.cpp

@@ -3129,8 +3129,7 @@ static bool packet_receive(rem_port* port, UCHAR* buffer, SSHORT buffer_length,
 	} // end scope
 #endif
 
-	port->port_rcv_packets++;
-	port->port_rcv_bytes += n;
+	port->bumpPhysStats(rem_port::RECEIVE, n);
 
 	*length = n;
 
@@ -3305,9 +3304,7 @@ static bool packet_send( rem_port* port, const SCHAR* buffer, SSHORT buffer_leng
 	} // end scope
 #endif
 
-	port->port_snd_packets++;
-	port->port_snd_bytes += buffer_length;
-
+	port->bumpPhysStats(rem_port::SEND, buffer_length);
 	return true;
 }
 

src/remote/os/win32/xnet.cpp

@@ -1991,8 +1991,8 @@ static bool_t xnet_read(RemoteXdr* xdrs)
 		{
 			// Client has written some data for us (server) to read
 
-			port->port_rcv_packets++;
-			port->port_rcv_bytes += xch->xch_length;
+			port->bumpPhysStats(rem_port::RECEIVE, xch->xch_length);
+			port->bumpLogBytes(rem_port::RECEIVE, xch->xch_length);	// XNET not calls REMOTE_inflate
 
 			xdrs->x_handy = xch->xch_length;
 			xdrs->x_private = xdrs->x_base;
@@ -2048,8 +2048,8 @@ static bool_t xnet_write(RemoteXdr* xdrs)
 	xch->xch_length = xdrs->x_private - xdrs->x_base;
 	if (SetEvent(xcc->xcc_event_send_channel_filled))
 	{
-		port->port_snd_packets++;
-		port->port_snd_bytes += xch->xch_length;
+		port->bumpPhysStats(rem_port::SEND, xch->xch_length);
+		port->bumpLogBytes(rem_port::SEND, xch->xch_length);	// XNET not calls REMOTE_deflate
 
 		xdrs->x_private = xdrs->x_base;
 		xdrs->x_handy = xch->xch_size;

src/remote/protocol.cpp

@@ -303,6 +303,9 @@ bool_t xdr_protocol(RemoteXdr* xdrs, PACKET* p)
 
 	const auto port = xdrs->x_public;
 
+	if (xdrs->x_op != XDR_FREE)
+		port->bumpLogPackets(xdrs->x_op == XDR_ENCODE ? rem_port::SEND : rem_port::RECEIVE);
+
 	switch (p->p_operation)
 	{
 	case op_reject:

src/remote/remote.cpp

@@ -1493,7 +1493,12 @@ bool REMOTE_inflate(rem_port* port, PacketReceive* packet_receive, UCHAR* buffer
 {
 #ifdef WIRE_COMPRESS_SUPPORT
 	if (!port->port_compressed)
-		return packet_receive(port, buffer, buffer_length, length);
+	{
+		const bool ret = packet_receive(port, buffer, buffer_length, length);
+		if (ret)
+			port->bumpLogBytes(rem_port::RECEIVE, *length);
+		return ret;
+	}
 
 	z_stream& strm = port->port_recv_stream;
 	strm.avail_out = buffer_length;
@@ -1565,16 +1570,22 @@ bool REMOTE_inflate(rem_port* port, PacketReceive* packet_receive, UCHAR* buffer
 	fprintf(stderr, "ZLib buffer %s\n", port->port_z_data ? "has data" : "is empty");
 #endif
 
+	port->bumpLogBytes(rem_port::RECEIVE, *length);
 	return true;
 #else
-	return packet_receive(port, buffer, buffer_length, length);
+	const bool ret = packet_receive(port, buffer, buffer_length, length);
+	if (ret)
+		port->bumpLogBytes(rem_port::RECEIVE, *length);
+	return ret;
 #endif
 }
 
 bool REMOTE_deflate(RemoteXdr* xdrs, ProtoWrite* proto_write, PacketSend* packet_send, bool flush)
 {
-#ifdef WIRE_COMPRESS_SUPPORT
 	rem_port* port = xdrs->x_public;
+	port->bumpLogBytes(rem_port::SEND, xdrs->x_private - xdrs->x_base);
+
+#ifdef WIRE_COMPRESS_SUPPORT
 	if (!(port->port_compressed && (port->port_flags & PORT_compressed)))
 		return proto_write(xdrs);
 

src/remote/remote.h

@@ -1144,10 +1144,100 @@ struct rem_port : public Firebird::GlobalStorage, public Firebird::RefCounted
 
 	UCharArrayAutoPtr	port_buffer;
 
+
+	enum io_direction_t {
+		NONE,
+		SEND,
+		RECEIVE
+	};
+
+private:
+	// packets over physical connection
 	FB_UINT64 port_snd_packets;
 	FB_UINT64 port_rcv_packets;
+	// protocol packets
+	FB_UINT64 port_out_packets;
+	FB_UINT64 port_in_packets;
+	// bytes over physical connection
 	FB_UINT64 port_snd_bytes;
 	FB_UINT64 port_rcv_bytes;
+	// bytes before/after compression
+	FB_UINT64 port_out_bytes;
+	FB_UINT64 port_in_bytes;
+	FB_UINT64 port_roundtrips;				// number of changes of IO direction from SEND to RECEIVE
+	io_direction_t port_io_direction;		// last direction of IO
+
+public:
+	void bumpPhysStats(io_direction_t direction, ULONG count)
+	{
+		fb_assert(direction != NONE);
+
+		if (direction == SEND)
+		{
+			port_snd_packets++;
+			port_snd_bytes += count;
+		}
+		else
+		{
+			port_rcv_packets++;
+			port_rcv_bytes += count;
+		}
+
+		if (direction != port_io_direction)
+		{
+			if (port_io_direction != NONE && direction == RECEIVE)
+				port_roundtrips++;
+			port_io_direction = direction;
+		}
+	}
+
+	void bumpLogBytes(io_direction_t direction, ULONG count)
+	{
+		fb_assert(direction != NONE);
+
+		if (direction == SEND)
+			port_out_bytes += count;
+		else
+			port_in_bytes += count;
+	}
+
+	void bumpLogPackets(io_direction_t direction)
+	{
+		fb_assert(direction != NONE);
+
+		if (direction == SEND)
+			port_out_packets++;
+		else
+			port_in_packets++;
+	}
+
+	FB_UINT64 getStatItem(UCHAR infoItem) const
+	{
+		switch (infoItem)
+		{
+		case fb_info_wire_snd_packets:
+			return port_snd_packets;
+		case fb_info_wire_rcv_packets:
+			return port_rcv_packets;
+		case fb_info_wire_out_packets:
+			return port_out_packets;
+		case fb_info_wire_in_packets:
+			return port_in_packets;
+		case fb_info_wire_snd_bytes:
+			return port_snd_bytes;
+		case fb_info_wire_rcv_bytes:
+			return port_rcv_bytes;
+		case fb_info_wire_out_bytes:
+			return port_out_bytes;
+		case fb_info_wire_in_bytes:
+			return port_in_bytes;
+		case fb_info_wire_roundtrips:
+			return port_roundtrips;
+		default:
+			return 0;
+		}
+	}
+
 
 #ifdef WIRE_COMPRESS_SUPPORT
 	z_stream port_send_stream, port_recv_stream;
@@ -1187,7 +1277,9 @@ struct rem_port : public Firebird::GlobalStorage, public Firebird::RefCounted
 		port_known_server_keys(getPool()), port_crypt_plugin(NULL),
 		port_client_crypt_callback(NULL), port_server_crypt_callback(NULL), port_crypt_name(getPool()),
 		port_replicator(NULL), port_buffer(FB_NEW_POOL(getPool()) UCHAR[rpt]),
-		port_snd_packets(0), port_rcv_packets(0), port_snd_bytes(0), port_rcv_bytes(0)
+		port_snd_packets(0), port_rcv_packets(0), port_out_packets(0), port_in_packets(0),
+		port_snd_bytes(0), port_rcv_bytes(0), port_out_bytes(0), port_in_bytes(0),
+		port_roundtrips(0), port_io_direction(NONE)
 	{
 		addRef();
 		memset(&port_linger, 0, sizeof port_linger);