struct connection_throttle
{
	double    max_bytes_per_second;
	double    interval_bytes;
	struct timeval  interval_start;
	struct timeval last_throttle;
	double  interval_stop;
	double  sleep_adjust;

};

struct connection
{
// ....

	struct
	{
		struct timeval start;
		struct 
		{
			uint64_t	traffic;
			struct connection_throttle throttle;
		} io_in,io_out;
	} stats;
}

int connection_throttle(EV_P_ struct connection *con, struct connection_throttle *thr)
{
	puts(__PRETTY_FUNCTION__);

	if ( thr->max_bytes_per_second == 0 )
		return 64*1024;

	struct timeval now;
//	gettimeofday(&g_evtest->now, NULL);
	memcpy(&now, &g_evtest->now,  sizeof(struct timeval));


	double delta = 0.; // time for this session 
	double expect = 0.;

	double last_throttle;
	last_throttle = (now.tv_sec - thr->last_throttle.tv_sec)*1e3 + (now.tv_usec - thr->last_throttle.tv_usec) / 1e3;
	if ( last_throttle > 1000 )
	{
		printf("resetting connection\n");
		connection_throttle_reset(thr);
	}
	memcpy(&thr->last_throttle, &now, sizeof(struct timeval));


	delta = (now.tv_sec - thr->interval_start.tv_sec)*1e3 + (now.tv_usec - thr->interval_start.tv_usec) / 1e3;
	expect = (thr->interval_bytes / thr->max_bytes_per_second) * 1000;

	printf("throttle: delta  %f expect %f\n", delta, expect);

	int bytes = 1;
	bytes = (delta+125)* thr->max_bytes_per_second / 1000;
	bytes -= thr->interval_bytes;

	if ( expect > delta )
	{
		double slp = expect - delta;

		if ( slp < 200 && bytes > 0 )
		{
			thr->sleep_adjust = slp;
			printf("throttle: discarding sleep do %i bytes\n", bytes);
			return bytes;
		}

		if ( &con->stats.io_in.throttle == thr )
		{
		} else
			if ( &con->stats.io_out.throttle == thr )
		{
			printf("throttle: io_out\n");
			ev_io_stop(EV_A_ &con->events.io_out);
			if ( !ev_is_active(&con->events.throttle_io_out_timeout) ) 
			{
				if ( slp < 200 )
					slp =200;
				ev_timer_init(&con->events.throttle_io_out_timeout, connection_throttle_io_out_timeout_cb, (slp+thr->sleep_adjust)/1000.0, 0.);
				ev_timer_start(EV_A_ &con->events.throttle_io_out_timeout);
			}
			return 0;
		}
	} else
	{
		bytes = (delta+250)* thr->max_bytes_per_second / 1000;
		bytes -= thr->interval_bytes;
		printf("throttle: can do %i bytes\n", bytes);
	}

	return bytes;
}

void connection_throttle_update(EV_P_ struct connection *con, struct connection_throttle *thr, int bytes)
{
	printf("%s con %p thr %p bytes %i\n",__PRETTY_FUNCTION__, con, thr, bytes);
	if ( bytes > 0 )
		thr->interval_bytes += bytes;

	if ( &con->stats.io_in.throttle == thr )
	{
		con->stats.io_in.traffic += bytes;
	} else
		if ( &con->stats.io_out.throttle == thr )
	{
		con->stats.io_out.traffic += bytes;
	}
}

void connection_throttle_io_in_timeout_cb(EV_P_ struct ev_timer *w, int revents)
{                                      
	puts(__PRETTY_FUNCTION__);
}                                      

void connection_throttle_io_out_timeout_cb(EV_P_ struct ev_timer *w, int revents)
{
	puts(__PRETTY_FUNCTION__);
	struct connection *con = w->data;
	ev_io_start(EV_A_ &con->events.io_out);
}

void connection_throttle_reset(struct connection_throttle *thr)
{
	thr->interval_bytes = 0;
	thr->sleep_adjust = 0;
//	gettimeofday(&thr->interval_start, NULL);
	memcpy(&thr->interval_start,  &g_evtest->now,  sizeof(struct timeval));
}

void connection_tcp_io_out_cb(EV_P_ struct ev_io *w, int revents)
{
	printf("%s loop %p watcher %p con %p\n",__PRETTY_FUNCTION__, EV_A_ w, w->data);
	struct connection *con = w->data;
	int send_throttle = connection_throttle(EV_A_ con, &con->stats.io_out.throttle);
	int send_size = MIN(con->tcp.io_out->len, send_throttle);

	if ( send_size == 0 )
		return;

	int size = send(con->socket, con->tcp.io_out->str, send_size, 0);

	if ( ev_is_active(&con->events.connect_timeout) )
		ev_timer_again(EV_A_  &con->events.connect_timeout);

	if ( size > 0 )
	{
		connection_throttle_update(EV_A_ con, &con->stats.io_out.throttle, size);
		g_string_erase(con->tcp.io_out, 0 , size);
		if ( con->tcp.io_out->len == 0 )
		{
			ev_io_stop(EV_A_ w);
			if ( con->tcp.state == connection_state_close )
				connection_tcp_disconnect(EV_A_ con);
		}
	} else
		if ( size == -1 )
	{
		if ( errno == EAGAIN || errno == EWOULDBLOCK )
		{

		} else
			if ( revents != 0 )
			connection_tcp_disconnect(EV_A_ con);
	}
}