path: root/3rdparty/openpgm-svn-r1085/pgm/examples/pgmping.cc

/* vim:ts=8:sts=8:sw=4:noai:noexpandtab
 *
 * Simple send/reply ping tool using the PGM transport.
 *
 * With no arguments, one message is sent per second.
 *
 * Copyright (c) 2006-2010 Miru Limited.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

/* c99 compatibility for c++ */
#define __STDC_LIMIT_MACROS

/* Must be first for Sun */
#include "ping.pb.h"

/* c99 compatibility for c++ */
#define __STDC_FORMAT_MACROS
#define restrict

#include <cerrno>
#include <clocale>
#include <csignal>
#include <cstdio>
#include <cstdlib>
#include <cstring>

#include <inttypes.h>
#include <math.h>
#include <unistd.h>
#include <time.h>
#include <sys/time.h>
#ifdef CONFIG_HAVE_EPOLL
#	include <sys/epoll.h>
#endif
#include <sys/types.h>
#ifndef _WIN32
#	include <netdb.h>
#	include <netinet/in.h>
#	include <sched.h>
#	include <sys/socket.h>
#	include <arpa/inet.h>
#endif
#include <glib.h>
#include <pgm/pgm.h>
#ifdef CONFIG_WITH_HTTP
#	include <pgm/http.h>
#endif
#ifdef CONFIG_WITH_SNMP
#	include <pgm/snmp.h>
#endif

/* PGM internal time keeper */
typedef pgm_time_t (*pgm_time_update_func)(void);
extern pgm_time_update_func pgm_time_update_now;
extern "C" {
	size_t pgm_pkt_offset (bool, sa_family_t);
}

/* example dependencies */
#include <pgm/backtrace.h>
#include <pgm/log.h>
#include <pgm/signal.h>


using namespace std;


/* globals */

static int		g_port = 0;
static const char*	g_network = "";
static int		g_udp_encap_port = 0;

static int		g_odata_rate = 0;
static int		g_odata_interval = 0;
static guint32		g_payload = 0;
static int		g_max_tpdu = 1500;
static int		g_max_rte = 16*1000*1000;
static int		g_sqns = 200;

static gboolean		g_use_pgmcc = FALSE;
static sa_family_t	g_pgmcc_family = 0;	/* 0 = disabled */

static gboolean		g_use_fec = FALSE;
static int		g_rs_k = 8;
static int		g_rs_n = 255;

static enum {
	PGMPING_MODE_SOURCE,
	PGMPING_MODE_RECEIVER,
	PGMPING_MODE_INITIATOR,
	PGMPING_MODE_REFLECTOR
}			g_mode = PGMPING_MODE_INITIATOR;

static pgm_sock_t*	g_sock = NULL;

/* stats */
static guint64		g_msg_sent = 0;
static guint64		g_msg_received = 0;
static pgm_time_t	g_interval_start = 0;
static pgm_time_t	g_latency_current = 0;
static guint64		g_latency_seqno = 0;
static guint64		g_last_seqno = 0;
static double		g_latency_total = 0.0;
static double		g_latency_square_total = 0.0;
static guint64		g_latency_count = 0;
static double		g_latency_max = 0.0;
#ifdef INFINITY
static double		g_latency_min = INFINITY;
#else
static double		g_latency_min = INT64_MAX;
#endif
static double		g_latency_running_average = 0.0;
static guint64		g_out_total = 0;
static guint64		g_in_total = 0;

static GMainLoop*	g_loop = NULL;
static GThread*		g_sender_thread = NULL;
static GThread*		g_receiver_thread = NULL;
static gboolean		g_quit;
#ifdef G_OS_UNIX
static int		g_quit_pipe[2];
static void on_signal (int, gpointer);
#else
static HANDLE		g_quit_event;
static BOOL on_console_ctrl (DWORD);
#endif

static gboolean on_startup (gpointer);
static gboolean on_shutdown (gpointer);
static gboolean on_mark (gpointer);

static void send_odata (void);
static int on_msgv (struct pgm_msgv_t*, size_t);

static gpointer sender_thread (gpointer);
static gpointer receiver_thread (gpointer);


G_GNUC_NORETURN static void
usage (const char* bin)
{
	fprintf (stderr, "Usage: %s [options]\n", bin);
	fprintf (stderr, "  -n <network>    : Multicast group or unicast IP address\n");
	fprintf (stderr, "  -s <port>       : IP port\n");
        fprintf (stderr, "  -p <port>       : Encapsulate PGM in UDP on IP port\n");
	fprintf (stderr, "  -d <seconds>    : Terminate transport after duration.\n");
	fprintf (stderr, "  -m <frequency>  : Number of message to send per second\n");
	fprintf (stderr, "  -o              : Send-only mode (default send & receive mode)\n");
	fprintf (stderr, "  -l              : Listen-only mode\n");
	fprintf (stderr, "  -e              : Relect mode\n");
        fprintf (stderr, "  -r <rate>       : Regulate to rate bytes per second\n");
	fprintf (stderr, "  -c              : Enable PGMCC\n");
        fprintf (stderr, "  -f <type>       : Enable FEC with either proactive or ondemand parity\n");
        fprintf (stderr, "  -K <k>          : Configure Reed-Solomon code (n, k)\n");
        fprintf (stderr, "  -N <n>\n");
        fprintf (stderr, "  -H              : Enable HTTP administrative interface\n");
        fprintf (stderr, "  -S              : Enable SNMP interface\n");
	exit (1);
}

int
main (
	int	argc,
	char   *argv[]
	)
{
	GError* err = NULL;
	pgm_error_t* pgm_err = NULL;
	gboolean enable_http = FALSE;
	gboolean enable_snmpx = FALSE;
	int timeout = 0;

	GOOGLE_PROTOBUF_VERIFY_VERSION;

	setlocale (LC_ALL, "");
	setenv ("PGM_TIMER", "GTOD", 1);
	setenv ("PGM_SLEEP", "USLEEP", 1);

	log_init ();
	g_message ("pgmping");

	g_thread_init (NULL);

	if (!pgm_init (&pgm_err)) {
		g_error ("Unable to start PGM engine: %s", pgm_err->message);
		pgm_error_free (pgm_err);
		return EXIT_FAILURE;
	}

/* parse program arguments */
	const char* binary_name = g_get_prgname();
	int c;
	while ((c = getopt (argc, argv, "s:n:p:m:old:r:cfeK:N:HSh")) != -1)
	{
		switch (c) {
		case 'n':	g_network = optarg; break;
		case 's':	g_port = atoi (optarg); break;
		case 'p':	g_udp_encap_port = atoi (optarg); break;
		case 'r':	g_max_rte = atoi (optarg); break;

		case 'c':	g_use_pgmcc = TRUE; break;

		case 'f':	g_use_fec = TRUE; break;
		case 'K':	g_rs_k = atoi (optarg); break;
		case 'N':	g_rs_n = atoi (optarg); break;

		case 'H':	enable_http = TRUE; break;
		case 'S':	enable_snmpx = TRUE; break;

		case 'm':	g_odata_rate = atoi (optarg);
				g_odata_interval = (1000 * 1000) / g_odata_rate; break;
		case 'd':	timeout = 1000 * atoi (optarg); break;

		case 'o':	g_mode = PGMPING_MODE_SOURCE; break;
		case 'l':	g_mode = PGMPING_MODE_RECEIVER; break;
		case 'e':	g_mode = PGMPING_MODE_REFLECTOR; break;

		case 'h':
		case '?': usage (binary_name);
		}
	}

	if (g_use_fec && ( !g_rs_k || !g_rs_n )) {
		g_error ("Invalid Reed-Solomon parameters.");
		usage (binary_name);
	}

#ifdef CONFIG_WITH_HTTP
	if (enable_http) {
		if (!pgm_http_init (PGM_HTTP_DEFAULT_SERVER_PORT, &pgm_err)) {
			g_error ("Unable to start HTTP interface: %s", pgm_err->message);
			pgm_error_free (pgm_err);
			pgm_shutdown ();
			return EXIT_FAILURE;
		}
	}
#endif
#ifdef CONFIG_WITH_SNMP
	if (enable_snmpx) {
		if (!pgm_snmp_init (&pgm_err)) {
			g_error ("Unable to start SNMP interface: %s", pgm_err->message);
			pgm_error_free (pgm_err);
#ifdef CONFIG_WITH_HTTP
			if (enable_http)
				pgm_http_shutdown ();
#endif
			pgm_shutdown ();
			return EXIT_FAILURE;
		}
	}
#endif

	g_loop = g_main_loop_new (NULL, FALSE);

	g_quit = FALSE;

/* setup signal handlers */
	signal (SIGSEGV, on_sigsegv);
#ifdef SIGHUP
	signal (SIGHUP,  SIG_IGN);
#endif
#ifdef G_OS_UNIX
	pipe (g_quit_pipe);
	pgm_signal_install (SIGINT,  on_signal, g_loop);
	pgm_signal_install (SIGTERM, on_signal, g_loop);
#else
	g_quit_event = CreateEvent (NULL, TRUE, FALSE, TEXT("QuitEvent"));
	SetConsoleCtrlHandler ((PHANDLER_ROUTINE)on_console_ctrl, TRUE);
#endif /* !G_OS_UNIX */

/* delayed startup */
	g_message ("scheduling startup.");
	g_timeout_add (0, (GSourceFunc)on_startup, g_loop);

	if (timeout) {
		g_message ("scheduling shutdown.");
		g_timeout_add (timeout, (GSourceFunc)on_shutdown, g_loop);
	}

/* dispatch loop */
	g_message ("entering main event loop ... ");
	g_main_loop_run (g_loop);

	g_message ("event loop terminated, cleaning up.");

/* cleanup */
	g_quit = TRUE;
#ifdef G_OS_UNIX
	const char one = '1';
	write (g_quit_pipe[1], &one, sizeof(one));
	if (PGMPING_MODE_SOURCE == g_mode || PGMPING_MODE_INITIATOR == g_mode)
		g_thread_join (g_sender_thread);
	g_thread_join (g_receiver_thread);
	close (g_quit_pipe[0]);
	close (g_quit_pipe[1]);
#else
	SetEvent (g_quit_event);
	if (PGMPING_MODE_SOURCE == g_mode || PGMPING_MODE_INITIATOR == g_mode)
		g_thread_join (g_sender_thread);
	g_thread_join (g_receiver_thread);
	CloseHandle (g_quit_event);
#endif

	g_main_loop_unref (g_loop);
	g_loop = NULL;

	if (g_sock) {
		g_message ("closing PGM socket.");
		pgm_close (g_sock, TRUE);
		g_sock = NULL;
	}

#ifdef CONFIG_WITH_HTTP
	if (enable_http)
		pgm_http_shutdown();
#endif
#ifdef CONFIG_WITH_SNMP
	if (enable_snmpx)
		pgm_snmp_shutdown();
#endif

	google::protobuf::ShutdownProtobufLibrary();

	g_message ("PGM engine shutdown.");
	pgm_shutdown ();
	g_message ("finished.");
	return EXIT_SUCCESS;
}

#ifdef G_OS_UNIX
static
void
on_signal (
	int		signum,
	gpointer	user_data
	)
{
	GMainLoop* loop = (GMainLoop*)user_data;
	g_message ("on_signal (signum:%d user-data:%p)",
		   signum, user_data);
	g_main_loop_quit (loop);
}
#else
static
BOOL
on_console_ctrl (
	DWORD		dwCtrlType
	)
{
	g_message ("on_console_ctrl (dwCtrlType:%lu)", (unsigned long)dwCtrlType);
	g_main_loop_quit (g_loop);
	return TRUE;
}
#endif /* !G_OS_UNIX */

static
gboolean
on_shutdown (
	gpointer	user_data
	)
{
	GMainLoop* loop = (GMainLoop*)user_data;
	g_message ("on_shutdown (user-data:%p)", user_data);
	g_main_loop_quit (loop);
	return FALSE;
}

static
gboolean
on_startup (
	gpointer	user_data
	)
{
	GMainLoop* loop = (GMainLoop*)user_data;
	struct pgm_addrinfo_t* res = NULL;
	GError* err = NULL;
	pgm_error_t* pgm_err = NULL;
	sa_family_t sa_family = AF_UNSPEC;

	g_message ("startup.");

/* parse network parameter into transport address structure */
	if (!pgm_getaddrinfo (g_network, NULL, &res, &pgm_err)) {
		g_error ("parsing network parameter: %s", pgm_err->message);
		goto err_abort;
	}

	sa_family = res->ai_send_addrs[0].gsr_group.ss_family;
	if (g_use_pgmcc)
		g_pgmcc_family = sa_family;

	if (g_udp_encap_port) {
		g_message ("create PGM/UDP socket.");
		if (!pgm_socket (&g_sock, sa_family, SOCK_SEQPACKET, IPPROTO_UDP, &pgm_err)) {
			g_error ("socket: %s", pgm_err->message);
			goto err_abort;
		}
		pgm_setsockopt (g_sock, PGM_UDP_ENCAP_UCAST_PORT, &g_udp_encap_port, sizeof(g_udp_encap_port));
		pgm_setsockopt (g_sock, PGM_UDP_ENCAP_MCAST_PORT, &g_udp_encap_port, sizeof(g_udp_encap_port));
	} else {
		g_message ("create PGM/IP socket.");
		if (!pgm_socket (&g_sock, sa_family, SOCK_SEQPACKET, IPPROTO_PGM, &pgm_err)) {
			g_error ("socket: %s", pgm_err->message);
			goto err_abort;
		}
	}

/* Use RFC 2113 tagging for PGM Router Assist */
	{
		const int no_router_assist = 0;
		pgm_setsockopt (g_sock, PGM_IP_ROUTER_ALERT, &no_router_assist, sizeof(no_router_assist));
	}

	pgm_drop_superuser();

/* set PGM parameters */
	if (PGMPING_MODE_SOURCE == g_mode ||
	    PGMPING_MODE_INITIATOR == g_mode ||
	    PGMPING_MODE_REFLECTOR == g_mode)
	{
		const int send_only	  = PGMPING_MODE_SOURCE == g_mode ? 1 : 0,
			  ambient_spm	  = pgm_secs (30),
			  heartbeat_spm[] = { pgm_msecs (100),
					      pgm_msecs (100),
					      pgm_msecs (100),
					      pgm_msecs (100),
					      pgm_msecs (1300),
					      pgm_secs  (7),
					      pgm_secs  (16),
					      pgm_secs  (25),
					      pgm_secs  (30) };

		pgm_setsockopt (g_sock, PGM_SEND_ONLY, &send_only, sizeof(send_only));
		pgm_setsockopt (g_sock, PGM_MTU, &g_max_tpdu, sizeof(g_max_tpdu));
		pgm_setsockopt (g_sock, PGM_TXW_SQNS, &g_sqns, sizeof(g_sqns));
		pgm_setsockopt (g_sock, PGM_TXW_MAX_RTE, &g_max_rte, sizeof(g_max_rte));
		pgm_setsockopt (g_sock, PGM_AMBIENT_SPM, &ambient_spm, sizeof(ambient_spm));
		pgm_setsockopt (g_sock, PGM_HEARTBEAT_SPM, &heartbeat_spm, sizeof(heartbeat_spm));
	}
	if (PGMPING_MODE_RECEIVER == g_mode ||
	    PGMPING_MODE_INITIATOR == g_mode ||
	    PGMPING_MODE_REFLECTOR == g_mode)
	{
		const int recv_only	   = PGMPING_MODE_RECEIVER == g_mode ? 1 : 0,
			  passive	   = 0,
			  peer_expiry	   = pgm_secs (300),
			  spmr_expiry	   = pgm_msecs (250),
			  nak_bo_ivl	   = pgm_msecs (50),
			  nak_rpt_ivl	   = pgm_msecs (200), //pgm_secs (2),
			  nak_rdata_ivl    = pgm_msecs (200), //pgm_secs (2),
			  nak_data_retries = 50,
			  nak_ncf_retries  = 50;

		pgm_setsockopt (g_sock, PGM_RECV_ONLY, &recv_only, sizeof(recv_only));
		pgm_setsockopt (g_sock, PGM_PASSIVE, &passive, sizeof(passive));
		pgm_setsockopt (g_sock, PGM_MTU, &g_max_tpdu, sizeof(g_max_tpdu));
		pgm_setsockopt (g_sock, PGM_RXW_SQNS, &g_sqns, sizeof(g_sqns));
		pgm_setsockopt (g_sock, PGM_PEER_EXPIRY, &peer_expiry, sizeof(peer_expiry));
		pgm_setsockopt (g_sock, PGM_SPMR_EXPIRY, &spmr_expiry, sizeof(spmr_expiry));
		pgm_setsockopt (g_sock, PGM_NAK_BO_IVL, &nak_bo_ivl, sizeof(nak_bo_ivl));
		pgm_setsockopt (g_sock, PGM_NAK_RPT_IVL, &nak_rpt_ivl, sizeof(nak_rpt_ivl));
		pgm_setsockopt (g_sock, PGM_NAK_RDATA_IVL, &nak_rdata_ivl, sizeof(nak_rdata_ivl));
		pgm_setsockopt (g_sock, PGM_NAK_DATA_RETRIES, &nak_data_retries, sizeof(nak_data_retries));
		pgm_setsockopt (g_sock, PGM_NAK_NCF_RETRIES, &nak_ncf_retries, sizeof(nak_ncf_retries));
	}

#ifdef I_UNDERSTAND_PGMCC_AND_FEC_ARE_NOT_SUPPORTED
/* PGMCC congestion control */
	if (g_use_pgmcc) {
		struct pgm_pgmccinfo_t pgmccinfo;
		pgmccinfo.ack_bo_ivl		= pgm_msecs (50);
		pgmccinfo.ack_c			= 75;
		pgmccinfo.ack_c_p		= 500;
		pgm_setsockopt (g_sock, PGM_USE_PGMCC, &pgmccinfo, sizeof(pgmccinfo));
	}

/* Reed Solomon forward error correction */
	if (g_use_fec) {
		struct pgm_fecinfo_t fecinfo; 
		fecinfo.block_size		= g_rs_n;
		fecinfo.proactive_packets	= 0;
		fecinfo.group_size		= g_rs_k;
		fecinfo.ondemand_parity_enabled	= TRUE;
		fecinfo.var_pktlen_enabled	= TRUE;
		pgm_setsockopt (g_sock, PGM_USE_FEC, &fecinfo, sizeof(fecinfo));
	}
#endif

/* create global session identifier */
	struct pgm_sockaddr_t addr;
	memset (&addr, 0, sizeof(addr));
	addr.sa_port = g_port ? g_port : DEFAULT_DATA_DESTINATION_PORT;
	addr.sa_addr.sport = DEFAULT_DATA_SOURCE_PORT;
	if (!pgm_gsi_create_from_hostname (&addr.sa_addr.gsi, &pgm_err)) {
		g_error ("creating GSI: %s", pgm_err->message);
		goto err_abort;
	}

/* assign socket to specified address */
	struct pgm_interface_req_t if_req;
	memset (&if_req, 0, sizeof(if_req));
	if_req.ir_interface = res->ai_recv_addrs[0].gsr_interface;
	if_req.ir_scope_id  = 0;
	if (AF_INET6 == sa_family) {
		struct sockaddr_in6 sa6;
		memcpy (&sa6, &res->ai_recv_addrs[0].gsr_group, sizeof(sa6));
		if_req.ir_scope_id = sa6.sin6_scope_id;
	}
	if (!pgm_bind3 (g_sock,
			&addr, sizeof(addr),
			&if_req, sizeof(if_req),	/* tx interface */
			&if_req, sizeof(if_req),	/* rx interface */
			&pgm_err))
	{
		g_error ("binding PGM socket: %s", pgm_err->message);
		goto err_abort;
	}

/* join IP multicast groups */
	for (unsigned i = 0; i < res->ai_recv_addrs_len; i++)
		pgm_setsockopt (g_sock, PGM_JOIN_GROUP, &res->ai_recv_addrs[i], sizeof(struct group_req));
	pgm_setsockopt (g_sock, PGM_SEND_GROUP, &res->ai_send_addrs[0], sizeof(struct group_req));
	pgm_freeaddrinfo (res);

/* set IP parameters */
	{
		const int nonblocking	 = 1,
			  multicast_loop = 0,
			  multicast_hops = 16,
			  dscp		 = 0x2e << 2;	/* Expedited Forwarding PHB for network elements, no ECN. */

		pgm_setsockopt (g_sock, PGM_MULTICAST_LOOP, &multicast_loop, sizeof(multicast_loop));
		pgm_setsockopt (g_sock, PGM_MULTICAST_HOPS, &multicast_hops, sizeof(multicast_hops));
		if (AF_INET6 != sa_family)
			pgm_setsockopt (g_sock, PGM_TOS, &dscp, sizeof(dscp));
		pgm_setsockopt (g_sock, PGM_NOBLOCK, &nonblocking, sizeof(nonblocking));
	}

	if (!pgm_connect (g_sock, &pgm_err)) {
		g_error ("connecting PGM socket: %s", pgm_err->message);
		goto err_abort;
	}

/* period timer to indicate some form of life */
// TODO: Gnome 2.14: replace with g_timeout_add_seconds()
	g_timeout_add (2 * 1000, (GSourceFunc)on_mark, NULL);

	if (PGMPING_MODE_SOURCE == g_mode || PGMPING_MODE_INITIATOR == g_mode)
	{
		g_sender_thread = g_thread_create_full (sender_thread,
							g_sock,
							0,
							TRUE,
							TRUE,
							G_THREAD_PRIORITY_NORMAL,
							&err);
		if (!g_sender_thread) {
			g_critical ("g_thread_create_full failed errno %i: \"%s\"", err->code, err->message);
			goto err_abort;
		}
	}

	{
		g_receiver_thread = g_thread_create_full (receiver_thread,
							  g_sock,
							  0,
							  TRUE,
							  TRUE,
							  G_THREAD_PRIORITY_HIGH,
							  &err);
		if (!g_receiver_thread) {
			g_critical ("g_thread_create_full failed errno %i: \"%s\"", err->code, err->message);
			goto err_abort;
		}
	}

	g_message ("startup complete.");
	return FALSE;

err_abort:
	if (NULL != g_sock) {
		pgm_close (g_sock, FALSE);
		g_sock = NULL;
	}
	if (NULL != res) {
		pgm_freeaddrinfo (res);
		res = NULL;
	}
	if (NULL != pgm_err) {
		pgm_error_free (pgm_err);
		pgm_err = NULL;
	}
	return FALSE;
}

static
gpointer
sender_thread (
	gpointer	user_data
	)
{
	pgm_sock_t* tx_sock = (pgm_sock_t*)user_data;
	example::Ping ping;
	string subject("PING.PGM.TEST.");
	char hostname[NI_MAXHOST + 1];
	const long payload_len = 1000;
	char payload[payload_len];
	gpointer buffer = NULL;
	guint64 latency, now, last;

#ifdef CONFIG_HAVE_EPOLL
	const long ev_len = 1;
	struct epoll_event events[ev_len];

	int efd_again = epoll_create (IP_MAX_MEMBERSHIPS);
	if (efd_again < 0) {
		g_error ("epoll_create failed errno %i: \"%s\"", errno, strerror(errno));
		g_main_loop_quit (g_loop);
		return NULL;
	}
/* Add write event to epoll domain in order to re-enable as required by return
 * value.  We use one-shot flag to disable ASAP, as we don't want such events
 * until triggered.
 */
	if (pgm_epoll_ctl (tx_sock, efd_again, EPOLL_CTL_ADD, EPOLLOUT | EPOLLONESHOT) < 0) {
		g_error ("pgm_epoll_ctl failed errno %i: \"%s\"", errno, strerror(errno));
		g_main_loop_quit (g_loop);
		return NULL;
	}
	struct epoll_event event;
	memset (&event, 0, sizeof(event));
	event.events = EPOLLIN;
	event.data.fd = g_quit_pipe[0];
	if (epoll_ctl (efd_again, EPOLL_CTL_ADD, g_quit_pipe[0], &event) < 0) {
		g_error ("epoll_ctl failed errno %i: \"%s\"", errno, strerror(errno));
		g_main_loop_quit (g_loop);
		return NULL;
	}
#elif defined(CONFIG_HAVE_POLL)
	int n_fds = 2;
	struct pollfd fds[ 2 + 1 ];
#endif /* !CONFIG_HAVE_EPOLL */

	gethostname (hostname, sizeof(hostname));
	subject.append(hostname);
	memset (payload, 0, sizeof(payload));

	ping.mutable_subscription_header()->set_subject (subject);
	ping.mutable_market_data_header()->set_msg_type (example::MarketDataHeader::MSG_VERIFY);
	ping.mutable_market_data_header()->set_rec_type (example::MarketDataHeader::PING);
	ping.mutable_market_data_header()->set_rec_status (example::MarketDataHeader::STATUS_OK);
	ping.set_time (last);

	last = now = pgm_time_update_now();
	do {
		if (g_msg_sent && g_latency_seqno + 1 == g_msg_sent)
			latency = g_latency_current;
		else
			latency = g_odata_interval;

		ping.set_seqno (g_msg_sent);
		ping.set_latency (latency);
		ping.set_payload (payload, sizeof(payload));

		const size_t header_size = pgm_pkt_offset (FALSE, g_pgmcc_family);
		const size_t apdu_size = ping.ByteSize();
		struct pgm_sk_buff_t* skb = pgm_alloc_skb (g_max_tpdu);
		pgm_skb_reserve (skb, header_size);
		pgm_skb_put (skb, apdu_size);

/* wait on packet rate limit */
		if ((last + g_odata_interval) > now) {
#ifndef _WIN32
			const unsigned int usec = g_odata_interval - (now - last);
			usleep (usec);
#else
			const DWORD msec = usecs_to_msecs (g_odata_interval - (now - last));
			Sleep (msec);
#endif
			now = pgm_time_update_now();
		}
		last += g_odata_interval;
		ping.set_time (now);
		ping.SerializeToArray (skb->data, skb->len);

		struct timeval tv;
		int timeout;
		size_t bytes_written;
		int status;
again:
		status = pgm_send_skbv (tx_sock, &skb, 1, TRUE, &bytes_written);
		switch (status) {
/* rate control */
		case PGM_IO_STATUS_RATE_LIMITED:
		{
			socklen_t optlen = sizeof (tv);
			pgm_getsockopt (tx_sock, PGM_RATE_REMAIN, &tv, &optlen);
			timeout = (tv.tv_sec * 1000) + ((tv.tv_usec + 500) / 1000);
/* busy wait under 2ms */
			if (timeout < 2) timeout = 0;
#ifdef CONFIG_HAVE_EPOLL
			const int ready = epoll_wait (efd_again, events, G_N_ELEMENTS(events), timeout /* ms */);
#elif defined(CONFIG_HAVE_POLL)
			memset (fds, 0, sizeof(fds));
			fds[0].fd = g_quit_pipe[0];
			fds[0].events = POLLIN;
			pgm_poll_info (tx_sock, &fds[1], &n_fds, POLLIN);
			poll (fds, 1 + n_fds, timeout /* ms */);
#endif /* !CONFIG_HAVE_EPOLL */
			if (G_UNLIKELY(g_quit))
				break;
			goto again;
		}
/* congestion control */
		case PGM_IO_STATUS_CONGESTION:
/* kernel feedback */
		case PGM_IO_STATUS_WOULD_BLOCK:
		{
#ifdef CONFIG_HAVE_EPOLL
/* re-enable write event for one-shot */
			if (pgm_epoll_ctl (tx_sock, efd_again, EPOLL_CTL_MOD, EPOLLOUT | EPOLLONESHOT) < 0)
			{
				g_error ("pgm_epoll_ctl failed errno %i: \"%s\"", errno, strerror(errno));
				g_main_loop_quit (g_loop);
				return NULL;
			}
			const int ready = epoll_wait (efd_again, events, G_N_ELEMENTS(events), -1 /* ms */);
			if (G_UNLIKELY(g_quit))
				break;
#elif defined(CONFIG_HAVE_POLL)
			memset (fds, 0, sizeof(fds));
			fds[0].fd = g_quit_pipe[0];
			fds[0].events = POLLIN;
			pgm_poll_info (g_sock, &fds[1], &n_fds, POLLOUT);
			poll (fds, 1 + n_fds, -1 /* ms */);
#endif /* !CONFIG_HAVE_EPOLL */
			goto again;
		}
/* successful delivery */
		case PGM_IO_STATUS_NORMAL:
//			g_message ("sent payload: %s", ping.DebugString().c_str());
//			g_message ("sent %u bytes", (unsigned)bytes_written);
			break;
		default:
			g_warning ("pgm_send_skbv failed, status:%i", status);
			g_main_loop_quit (g_loop);
			return NULL;
		}
		g_out_total += bytes_written;
		g_msg_sent++;
	} while (!g_quit);

#ifdef CONFIG_HAVE_EPOLL
	close (efd_again);
#endif
	return NULL;
}

static
gpointer
receiver_thread (
	gpointer	data
	)
{
	pgm_sock_t* rx_sock = (pgm_sock_t*)data;
	const long iov_len = 20;
	struct pgm_msgv_t msgv[iov_len];
	pgm_time_t lost_tstamp = 0;
	pgm_tsi_t  lost_tsi;
	guint32	   lost_count = 0;

#ifdef CONFIG_HAVE_EPOLL
	const long ev_len = 1;
	struct epoll_event events[ev_len];

	int efd = epoll_create (IP_MAX_MEMBERSHIPS);
	if (efd < 0) {
		g_error ("epoll_create failed errno %i: \"%s\"", errno, strerror(errno));
		g_main_loop_quit (g_loop);
		return NULL;
	}
	if (pgm_epoll_ctl (rx_sock, efd, EPOLL_CTL_ADD, EPOLLIN) < 0) {
		g_error ("pgm_epoll_ctl failed errno %i: \"%s\"", errno, strerror(errno));
		g_main_loop_quit (g_loop);
		return NULL;
	}
	struct epoll_event event;
	memset (&event, 0, sizeof(event));
	event.events = EPOLLIN;
	if (epoll_ctl (efd, EPOLL_CTL_ADD, g_quit_pipe[0], &event) < 0) {
		g_error ("epoll_ctl failed errno %i: \"%s\"", errno, strerror(errno));
		g_main_loop_quit (g_loop);
		return NULL;
	}
#elif defined(CONFIG_HAVE_POLL)
	int n_fds = 3;
	struct pollfd fds[ 3 + 1 ];
#endif /* !CONFIG_HAVE_EPOLL */

	memset (&lost_tsi, 0, sizeof(lost_tsi));

	do {
		struct timeval tv;
		int timeout;
		size_t len;
		pgm_error_t* pgm_err = NULL;
		const int status = pgm_recvmsgv (rx_sock,
					         msgv,
					         G_N_ELEMENTS(msgv),
					         MSG_ERRQUEUE,
					         &len,
					         &pgm_err);
		if (lost_count) {
			pgm_time_t elapsed = pgm_time_update_now() - lost_tstamp;
			if (elapsed >= pgm_secs(1)) {
				g_warning ("pgm data lost %" G_GUINT32_FORMAT " packets detected from %s",
						lost_count, pgm_tsi_print (&lost_tsi));
				lost_count = 0;
			}
		}

		switch (status) {
		case PGM_IO_STATUS_NORMAL:
//			g_message ("recv %u bytes", (unsigned)len);
			on_msgv (msgv, len);
			break;
		case PGM_IO_STATUS_TIMER_PENDING:
			{
				socklen_t optlen = sizeof (tv);
				pgm_getsockopt (g_sock, PGM_TIME_REMAIN, &tv, &optlen);
			}
			goto block;
		case PGM_IO_STATUS_RATE_LIMITED:
			{
				socklen_t optlen = sizeof (tv);
				pgm_getsockopt (g_sock, PGM_RATE_REMAIN, &tv, &optlen);
			}
/* fall through */
		case PGM_IO_STATUS_WOULD_BLOCK:
block:
			timeout = PGM_IO_STATUS_WOULD_BLOCK == status ? -1 : ((tv.tv_sec * 1000) + (tv.tv_usec / 1000));
/* busy wait under 2ms */
			if (timeout > 0 && timeout < 2) timeout = 0;
#ifdef CONFIG_HAVE_EPOLL
			epoll_wait (efd, events, G_N_ELEMENTS(events), timeout /* ms */);
#elif defined(CONFIG_HAVE_POLL)
			memset (fds, 0, sizeof(fds));
			fds[0].fd = g_quit_pipe[0];
			fds[0].events = POLLIN;
			pgm_transport_poll_info (g_transport, &fds[1], &n_fds, POLLIN);
			poll (fds, 1 + n_fds, timeout /* ms */);
#endif /* !CONFIG_HAVE_EPOLL */
			break;
		case PGM_IO_STATUS_RESET:
		{
			struct pgm_sk_buff_t* skb = msgv[0].msgv_skb[0];
			lost_tstamp = skb->tstamp;
			if (pgm_tsi_equal (&skb->tsi, &lost_tsi))
				lost_count += skb->sequence;
			else {
				lost_count = skb->sequence;
				memcpy (&lost_tsi, &skb->tsi, sizeof(pgm_tsi_t));
			}
			pgm_free_skb (skb);
			break;
		}
		default:
			if (pgm_err) {
				g_warning ("%s", pgm_err->message);
				pgm_error_free (pgm_err);
				pgm_err = NULL;
			}
			break;
		}
	} while (!g_quit);

#ifdef CONFIG_HAVE_EPOLL
	close (efd);
#endif
	return NULL;
}

static
int
on_msgv (
	struct pgm_msgv_t*	msgv,		/* an array of msgvs */
	size_t			len
	)
{
	example::Ping ping;
	guint i = 0;
	static pgm_time_t last_time = pgm_time_update_now();

	while (len)
	{
		const struct pgm_sk_buff_t* pskb = msgv[i].msgv_skb[0];
		gsize apdu_len = 0;
		for (unsigned j = 0; j < msgv[i].msgv_len; j++)
			apdu_len += msgv[i].msgv_skb[j]->len;

		if (PGMPING_MODE_REFLECTOR == g_mode)
		{
			int status;
again:
			status = pgm_send (g_sock, pskb->data, pskb->len, NULL);
			switch (status) {
			case PGM_IO_STATUS_RATE_LIMITED:
			case PGM_IO_STATUS_CONGESTION:
			case PGM_IO_STATUS_WOULD_BLOCK:
/* busy wait always as reflector */
				goto again;

			case PGM_IO_STATUS_NORMAL:
				break;

			default:
				g_warning ("pgm_send_skbv failed");
				g_main_loop_quit (g_loop);
				return 0;
			}
			goto next_msg;
		}

/* only parse first fragment of each apdu */
		if (!ping.ParseFromArray (pskb->data, pskb->len))
			goto next_msg;
//		g_message ("payload: %s", ping.DebugString().c_str());

		{
			const pgm_time_t send_time	= ping.time();
			const pgm_time_t recv_time	= pskb->tstamp;
			const guint64 seqno		= ping.seqno();
			const guint64 latency		= ping.latency();

			if (seqno < g_latency_seqno) {
				g_message ("seqno replay?");
				goto next_msg;
			}

			g_in_total += pskb->len;
			g_msg_received++;

/* handle ping */
			const pgm_time_t now = pgm_time_update_now();
			if (send_time > now)
				g_warning ("send time %" PGM_TIME_FORMAT " newer than now %" PGM_TIME_FORMAT,
					   send_time, now);
			if (recv_time > now)
				g_warning ("recv time %" PGM_TIME_FORMAT " newer than now %" PGM_TIME_FORMAT,
					   recv_time, now);
			if (send_time >= recv_time){
				g_message ("timer mismatch, send time = recv time + %.3f ms (last time + %.3f ms)",
					   pgm_to_msecsf(send_time - recv_time),
					   pgm_to_msecsf(last_time - send_time));
				goto next_msg;
			}
			g_latency_current	= pgm_to_secs(recv_time - send_time);
			g_latency_seqno		= seqno;

			const double elapsed    = pgm_to_usecsf (recv_time - send_time);
			g_latency_total	       += elapsed;
			g_latency_square_total += elapsed * elapsed;

			if (elapsed > g_latency_max)
				g_latency_max = elapsed;
			if (elapsed < g_latency_min)
				g_latency_min = elapsed;

			g_latency_running_average += elapsed;
			g_latency_count++;
			last_time = recv_time;
		}

/* move onto next apdu */
next_msg:
		i++;
		len -= apdu_len;
	}

	return 0;
}

/* idle log notification
 */

static
gboolean
on_mark (
	G_GNUC_UNUSED gpointer data
	)
{
	const pgm_time_t now = pgm_time_update_now ();
	const double interval = pgm_to_secsf(now - g_interval_start);
	g_interval_start = now;

/* receiving a ping */
	if (g_latency_count)
	{
		const double average = g_latency_total / g_latency_count;
		const double variance = g_latency_square_total / g_latency_count
					- average * average;
		const double standard_deviation = sqrt (variance);

		if (g_latency_count < 10)
		{
			if (average < 1000.0)
				g_message ("seqno=%" G_GUINT64_FORMAT " time=%.01f us",
						g_latency_seqno, average);
			else
				g_message ("seqno=%" G_GUINT64_FORMAT " time=%.01f ms",
						g_latency_seqno, average / 1000);
		}
		else
		{
			double seq_rate = (g_latency_seqno - g_last_seqno) / interval;
			double out_rate = g_out_total * 8.0 / 1000000.0 / interval;
			double  in_rate = g_in_total  * 8.0 / 1000000.0 / interval;
			if (g_latency_min < 1000.0)
				g_message ("s=%.01f avg=%.01f min=%.01f max=%.01f stddev=%0.1f us o=%.2f i=%.2f mbit",
					seq_rate, average, g_latency_min, g_latency_max, standard_deviation, out_rate, in_rate);
			else
				g_message ("s=%.01f avg=%.01f min=%.01f max=%.01f stddev=%0.1f ms o=%.2f i=%.2f mbit",
					seq_rate, average / 1000, g_latency_min / 1000, g_latency_max / 1000, standard_deviation / 1000, out_rate, in_rate);
		}

/* reset interval counters */
		g_latency_total		= 0.0;
		g_latency_square_total	= 0.0;
		g_latency_count		= 0;
		g_last_seqno		= g_latency_seqno;
#ifdef INFINITY
		g_latency_min		= INFINITY;
#else
		g_latency_min		= INT64_MAX;
#endif
		g_latency_max		= 0.0;
		g_out_total		= 0;
		g_in_total		= 0;
	}

	return TRUE;
}

/* eof */