#define _THREAD_SAFE


#include <unistd.h>
#include <sys/types.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <sys/poll.h>
#include <sys/pollset.h>
#include <fcntl.h>

int M_pipefd[2];

volatile int M_msg_proc = 0;
volatile int M_msg_wasted = 0;
pthread_mutex_t M_mutex = PTHREAD_MUTEX_INITIALIZER;


static void * waiter_thread_start(void *arg)
{
	char buf[1];
	pollset_t  ps;
	struct poll_ctl cl;
	struct pollfd polldata_array;
	int fd;
	struct pollfd fds[1];

	cl.cmd=PS_ADD;
	cl.events=POLLIN;
	cl.fd=M_pipefd[0];

	ps = pollset_create(1);
	
	if (0!=pollset_ctl(ps, &cl, 1))
	{
		perror("pollset_ctl");
		exit(-1);
	}
	
	while (pollset_poll(ps, &polldata_array, 1, 999999) > 0)
	{
		
		 if (read( M_pipefd[0], &buf, sizeof(buf))>=0)
		 {
			pthread_mutex_lock(&M_mutex);
			M_msg_proc++;
			pthread_mutex_unlock(&M_mutex);
		 }
		 else if (errno!=EAGAIN && errno!=EWOULDBLOCK)
		 {
			 perror("Failed to read");
			 exit(-1);
		 }
		 else
		 {
			pthread_mutex_lock(&M_mutex);
			M_msg_wasted++;
			pthread_mutex_unlock(&M_mutex); 
		 }
		 
	}
	
	perror( "client: read/poll failed" );
}

#define MAX_THREADS 5000

int main(int argc, char **argv) 
{
	int i;
	pthread_t tinfo[MAX_THREADS];
	int num_threads=1;
	int num_msg=1000000;
	time_t timer;
	time_t timer2;
	char buffer[26];
	struct tm* tm_info;
	/*int fd;*/
	int flags;
	
	if (argc<3)
	{
		fprintf(stderr, "Expected: %s <number_of_threads> <msg_num>\n", argv[0]);
		exit(-1);
	}

	if (-1==pipe(M_pipefd))
	{
		perror("pipe");
	}
	
	
	flags = fcntl(M_pipefd[0], F_GETFL, 0);
	if (fcntl(M_pipefd[0], F_SETFL, flags | O_NONBLOCK))
	{
		perror("fcnt");
	}
	
	num_threads=atoi(argv[1]);
	num_msg=atoi(argv[2]);
	
	if (num_threads>MAX_THREADS)
	{
		num_threads=MAX_THREADS;
	}

	for (i=0; i<num_threads; i++)
	{
		if (0!=pthread_create(&tinfo[i], NULL, &waiter_thread_start, NULL))
		{
			fprintf(stderr, "Failed to start thread %d: %s\n", i, strerror(errno));
		}
	}
	

	fprintf(stderr, "Wait 5s for threads to start... (num_threads=%d num_msg=%d)\n",
		num_threads, num_msg);
	sleep(5);

	timer = time(NULL);
	tm_info = localtime(&timer);
	strftime(buffer, 26, "%Y-%m-%d %H:%M:%S", tm_info);

	printf( "server: FD = R: %d W: %d\n", M_pipefd[0], M_pipefd[1]);
	printf( "server: START %s\n", buffer );

	for (i=0; i<num_msg; i++)
	{
		char buf[1]; /* send some random byte ... */
		
		if (-1==write(M_pipefd[1], buf, 1))
		{
			perror( "server: write failed:" );
			exit( 3 );
		}
	}

	timer2 = time(NULL);
	tm_info = localtime(&timer2);
	strftime(buffer, 26, "%Y-%m-%d %H:%M:%S", tm_info);
	fprintf( stderr, "server: STOP %s DELTA sec: %ld\n", buffer, timer2-timer);

	fprintf( stderr, "server: Messages %d sent successfully\n", num_msg);

	while (M_msg_proc!=num_msg)
	{
		fprintf( stderr, "server: waiting 1s... M_msg_proc=%d, num_msg=%d\n", 
			 M_msg_proc, num_msg);
		sleep( 1 );
	}
	
	fprintf( stderr, "server: Messages %d\n", num_msg);

	timer = time(NULL);
	tm_info = localtime(&timer);
	strftime(buffer, 26, "%Y-%m-%d %H:%M:%S", tm_info);
	printf( "server: COMPLETED %s WASTED WAEKUPS: %d\n", buffer, M_msg_wasted);


	printf( "server: done waiting. remove threads...\n" );

	for (i=0; i<num_threads; i++)
	{
		pthread_cancel(tinfo[i]);
	}
	
	sleep(2);

	printf( "server: done waiting. remove pipes...\n" );
	close(M_pipefd[0]);
	close(M_pipefd[1]);
	
}