/*
 * Should receive timeouts only
 */
static void *event_dispatcher(void *arg)
{
	odp_event_t ev;

	(void)arg;

	if (ofp_init_local()) {
		OFP_ERR("Error: OFP local init failed.\n");
		return NULL;
	}

	while (1) {
		ev = odp_schedule(NULL, ODP_SCHED_WAIT);

		if (ev == ODP_EVENT_INVALID)
			continue;

		if (odp_event_type(ev) == ODP_EVENT_TIMEOUT) {
			ofp_timer_handle(ev);
			continue;
		}

		OFP_ERR("Error: unexpected event type: %u\n",
			  odp_event_type(ev));

		odp_buffer_free(odp_buffer_from_event(ev));
	}

	/* Never reached */
	return NULL;
}

static odp_cos_t build_cos_w_queue(const char *name)
{
	odp_cos_t cos;
	odp_queue_t queue_cos;
	odp_queue_param_t qparam;
	odp_cls_cos_param_t cos_param;

	odp_queue_param_init(&qparam);
	qparam.type = ODP_QUEUE_TYPE_SCHED;
	qparam.sched.prio  = ODP_SCHED_PRIO_DEFAULT;
	qparam.sched.sync  = ODP_SCHED_SYNC_ATOMIC;
	qparam.sched.group = ODP_SCHED_GROUP_ALL;

	queue_cos = odp_queue_create(name, &qparam);
	if (queue_cos == ODP_QUEUE_INVALID) {
		OFP_ERR("Failed to create queue\n");
		return ODP_COS_INVALID;
	}

	odp_cls_cos_param_init(&cos_param);
	cos_param.queue = queue_cos;
	cos_param.pool = odp_pool_lookup(SHM_PKT_POOL_NAME);
	cos = odp_cls_cos_create(name, &cos_param);
	if (cos == ODP_COS_INVALID) {
		OFP_ERR("Failed to create COS");
		odp_cos_destroy(cos);
		odp_queue_destroy(queue_cos);
		return ODP_COS_INVALID;
	}

	return cos;
}

int ofp_shared_memory_term_global(void)
{
	int n;
	int ret = 0;

	for (n = 0; n < OFP_SHM_BLOCKS_MAX; n++) {
		if (shm && shm->block[n].valid && shm->block[n].allocated) {
			OFP_ERR("Shared memory block \"%s\" not freed,"
				"cannot free common shared memory",
				shm->block[n].name);
			return -1;
		}
	}

	if (ofp_shared_memory_free_raw(SHM_NAME_COMMON)) {
		OFP_ERR("Freeing shared memory failed (name: %s)",
			SHM_NAME_COMMON);
		ret = -1;
	}

	if (ofp_shared_memory_free_raw(SHM_NAME_INTERNAL)) {
		OFP_ERR("Freeing shared memory failed (name: %s)",
			SHM_NAME_INTERNAL);
		ret = -1;
	}
	shm = NULL;
	return ret;
}

static void *preallocated_alloc(const char *name, uint64_t size)
{
	struct ofp_shm_block *block;

	if (!shm) {
		OFP_DBG_SHM("Not initialized");
		return NULL;
	}

	block = ofp_shm_block_find(name);
	if (!block) {
		OFP_DBG_SHM("Not found in the preallocated memory");
		return NULL;
	}
	if (!shm->shared_memory) {
		OFP_ERR("Allocation of preallocated shared memory before "
			"preallocation phase has finished. Name: %s", name);
		return NULL;
	}
	if (block->allocated) {
		OFP_ERR("Shared memory (name: %s) already allocated", name);
		return NULL;
	}
	if (block->size < size) {
		OFP_WARN("Shared memory allocation (name: %s) larger than "
			 "the corresponding preallocation: "
			 "(%" PRIu64 " > %" PRIu64 ")",
			name, size, block->size);
		return NULL;
	}
	block->allocated = 1;

	return &shm->shared_memory[block->offset];
}

uma_zone_t ofp_uma_pool_create(const char *name, int nitems, int size)
{
	odp_pool_param_t pool_params;
	odp_pool_t pool;
	uma_zone_t zone;

	pool_params.buf.size  = size + sizeof(struct uma_pool_metadata);
	pool_params.buf.align = 0;
	pool_params.buf.num   = nitems;
	pool_params.type      = ODP_POOL_BUFFER;

	OFP_INFO("Creating pool '%s', nitems=%d size=%d total=%d",
		 name, pool_params.buf.num, pool_params.buf.size,
		 pool_params.buf.num * pool_params.buf.size);

	if (shm->num_pools >= OFP_NUM_UMA_POOLS) {
		OFP_ERR("Exceeded max number (%d) of pools",
			OFP_NUM_UMA_POOLS);
		return OFP_UMA_ZONE_INVALID;
	}
	pool = ofp_pool_create(name, &pool_params);
	if (pool == ODP_POOL_INVALID) {
		OFP_ERR("odp_pool_create failed");
		return OFP_UMA_ZONE_INVALID;
	}

	zone = shm->num_pools++;
	shm->pools[zone] = pool;

	return zone;
}

int send_ip6_udp_any(int fd)
{
	struct ofp_sockaddr_in6 addr = {0};
	const char *buf = "socket_test";

	addr.sin6_len = sizeof(struct ofp_sockaddr_in6);
	addr.sin6_family = OFP_AF_INET6;
	addr.sin6_port = odp_cpu_to_be_16(TEST_PORT + 1);
	addr.sin6_addr = ofp_in6addr_any;

	if (ofp_connect(fd, (const struct ofp_sockaddr *)&addr,
		sizeof(struct ofp_sockaddr_in6)) == -1) {
		OFP_ERR("Faild to connect socket (errno = %d)\n",
			ofp_errno);
		return -1;
	}

	if (ofp_send(fd, buf, strlen(buf), 0) == -1) {
		OFP_ERR("Faild to send data(errno = %d)\n", ofp_errno);
		return -1;
	}

	OFP_INFO("Data (%s) sent successfully.\n", buf);

	OFP_INFO("SUCCESS.\n");
	return 0;
}

static void *perf_client(void *arg)
{
	(void) arg;

#if ODP_VERSION < 106
	if (odp_init_local(ODP_THREAD_CONTROL) != 0) {
		OFP_ERR("Error: ODP local init failed.\n");
		return NULL;
	}
#endif
	if (ofp_init_local()) {
		OFP_ERR("Error: OFP local init failed.\n");
		return NULL;
	}

	ofp_set_stat_flags(OFP_STAT_COMPUTE_PERF);

	while (1) {
		struct ofp_perf_stat *ps = ofp_get_perf_statistics();
		printf ("Mpps:%4.3f\n", ((float)ps->rx_fp_pps)/1000000);
		usleep(1000000UL);
	}

	return NULL;
}

static int init_suite(void)
{
	odp_pool_param_t pool_params;
	ofp_pkt_hook pkt_hook[OFP_HOOK_MAX];
	odp_pool_t pool;
	odph_linux_thr_params_t thr_params;
	odp_instance_t instance;

	/* Init ODP before calling anything else */
	if (odp_init_global(&instance, NULL, NULL)) {
		OFP_ERR("Error: ODP global init failed.\n");
		return -1;
	}

	/* Init this thread */
	if (odp_init_local(instance, ODP_THREAD_CONTROL)) {
		OFP_ERR("Error: ODP local init failed.\n");
		return -1;
	}

	memset(pkt_hook, 0, sizeof(pkt_hook));

	pool_params.pkt.seg_len = SHM_PKT_POOL_BUFFER_SIZE;
	pool_params.pkt.len     = SHM_PKT_POOL_BUFFER_SIZE;
	pool_params.pkt.num     = SHM_PKT_POOL_SIZE / SHM_PKT_POOL_BUFFER_SIZE;
	pool_params.type        = ODP_POOL_PACKET;

	(void) ofp_init_pre_global(pool_name, &pool_params, pkt_hook, &pool,
				   ARP_AGE_INTERVAL, ARP_ENTRY_TIMEOUT);

	/*
	 * Start a packet processing thread to service timer events.
	 */
	odp_atomic_store_u32(&still_running, 1);

	odp_cpumask_t cpumask;
	odp_cpumask_zero(&cpumask);
	odp_cpumask_set(&cpumask, 0x1);
	thr_params.start = pp_thread;
	thr_params.arg = NULL;
	thr_params.thr_type = ODP_THREAD_WORKER;
	thr_params.instance = instance;
	odph_linux_pthread_create(&pp_thread_handle,
				&cpumask,
				&thr_params);

	return 0;
}

/*
 * Append address and data, and optionally, control (ancillary) data to the
 * receive queue of a socket.  If present, m0 must include a packet header
 * with total length.  Returns 0 if no space in sockbuf or insufficient
 * mbufs.
 */
int
ofp_sbappendaddr_locked(struct sockbuf *sb,
		    odp_packet_t pkt, odp_packet_t control)
{
	SOCKBUF_LOCK_ASSERT(sb);

	if (control != ODP_PACKET_INVALID)
		odp_packet_free(control);

	sb->sb_mb[sb->sb_put++] = pkt;
	if (sb->sb_put >= SOCKBUF_LEN)
		sb->sb_put = 0;

	if (sb->sb_put == sb->sb_get) {
		sb->sb_put--;
		if (sb->sb_put < 0)
			sb->sb_put = SOCKBUF_LEN-1;
		OFP_ERR("Buffers full, sb_get=%d max_num=%d",
			  sb->sb_get, SOCKBUF_LEN);
		return 0;
	}

	sballoc(sb, pkt);
	return (1);
}

void end_suite(void)
{
	if (fd_thread1 != -1) {
		if (ofp_close(fd_thread1) == -1)
			OFP_ERR("Faild to close socket 1 (errno = %d)\n",
				ofp_errno);
		fd_thread1 = -1;
	}

	if (fd_thread2 != -1) {
		if (ofp_close(fd_thread2) == -1)
			OFP_ERR("Faild to close socket 1 (errno = %d)\n",
				ofp_errno);
		fd_thread2 = -1;
	}
}

static inline int accept_connection(int serv_fd)
{
	int tmp_fd, i;
	struct ofp_sockaddr_in caller;
	unsigned int alen = sizeof(caller);

	if ((tmp_fd = ofp_accept(serv_fd, (struct ofp_sockaddr *)&caller, &alen)) > 0) {
		OFP_INFO("accept fd=%d", tmp_fd);

		for (i = 0; i < NUM_CONNECTIONS; i++)
			if (connections[i].fd == 0)
				break;

		if (i >= NUM_CONNECTIONS) {
			OFP_ERR("Node cannot accept new connections!");
			ofp_close(tmp_fd);
			return -1;
		}

#if 0
		struct ofp_linger so_linger;
		so_linger.l_onoff = 1;
		so_linger.l_linger = 0;
		int r1 = ofp_setsockopt(tmp_fd,
					  OFP_SOL_SOCKET,
					  OFP_SO_LINGER,
					  &so_linger,
					  sizeof so_linger);
		if (r1) OFP_ERR("SO_LINGER failed!");
#endif
		struct ofp_timeval tv;
		tv.tv_sec = 3;
		tv.tv_usec = 0;
		int r2 = ofp_setsockopt(tmp_fd,
					  OFP_SOL_SOCKET,
					  OFP_SO_SNDTIMEO,
					  &tv,
					  sizeof tv);
		if (r2) OFP_ERR("SO_SNDTIMEO failed!");

		connections[i].fd = tmp_fd;
		connections[i].addr = caller.sin_addr.s_addr;
		connections[i].closed = FALSE;
	}

	return tmp_fd;
}

int init_udp_bind_any(int *pfd_thread1, int *pfd_thread2)
{
	struct ofp_sockaddr_in addr = {0};


	*pfd_thread1 = ofp_socket(OFP_AF_INET, OFP_SOCK_DGRAM,
			OFP_IPPROTO_UDP);
	if (*pfd_thread1 == -1) {
		OFP_ERR("Faild to create SEND socket (errno = %d)\n",
			ofp_errno);
		return -1;
	}

	addr.sin_len = sizeof(struct ofp_sockaddr_in);
	addr.sin_family = OFP_AF_INET;
	addr.sin_port = odp_cpu_to_be_16(TEST_PORT);
	addr.sin_addr.s_addr = OFP_INADDR_ANY;

	if (ofp_bind(*pfd_thread1, (const struct ofp_sockaddr *)&addr,
		sizeof(struct ofp_sockaddr_in)) == -1) {
		OFP_ERR("Faild to bind socket (errno = %d)\n",
			ofp_errno);
		return -1;
	}

	*pfd_thread2 = ofp_socket(OFP_AF_INET, OFP_SOCK_DGRAM,
			OFP_IPPROTO_UDP);
	if (*pfd_thread2 == -1) {
		OFP_ERR("Faild to create RCV socket (errno = %d)\n",
			ofp_errno);
		return -1;
	}

	addr.sin_len = sizeof(struct ofp_sockaddr_in);
	addr.sin_family = OFP_AF_INET;
	addr.sin_port = odp_cpu_to_be_16(TEST_PORT + 1);
	addr.sin_addr.s_addr = OFP_INADDR_ANY;

	if (ofp_bind(*pfd_thread2, (const struct ofp_sockaddr *)&addr,
		sizeof(struct ofp_sockaddr_in)) == -1) {
		OFP_ERR("Faild to bind socket (errno = %d)\n",
			ofp_errno);
		return -1;
	}

	return 0;
}

static void *pkt_io_recv(void *arg)
{
	odp_pktio_t pktio;
	odp_packet_t pkt, pkt_tbl[OFP_PKT_BURST_SIZE];
	int pkt_idx, pkt_cnt;
	struct pktio_thr_arg *thr_args;
	ofp_pkt_processing_func pkt_func;

	thr_args = arg;
	pkt_func = thr_args->pkt_func;

	if (odp_init_local(ODP_THREAD_WORKER)) {
		OFP_ERR("Error: ODP local init failed.\n");
		return NULL;
	}
	if (ofp_init_local()) {
		OFP_ERR("Error: OFP local init failed.\n");
		return NULL;
	}

	pktio = ofp_port_pktio_get(thr_args->port);

	OFP_DBG("PKT-IO receive starting on port: %d, pktio-id: %"PRIX64"\n",
		  thr_args->port, odp_pktio_to_u64(pktio));

	while (1) {
		pkt_cnt = odp_pktio_recv(pktio, pkt_tbl, OFP_PKT_BURST_SIZE);

		for (pkt_idx = 0; pkt_idx < pkt_cnt; pkt_idx++) {
			pkt = pkt_tbl[pkt_idx];

			if (odp_unlikely(odp_packet_has_error(pkt))) {
				OFP_DBG("Packet with error dropped.\n");
				odp_packet_free(pkt);
				continue;
			}

			ofp_packet_input(pkt, ODP_QUEUE_INVALID, pkt_func);
		}
#ifdef OFP_SEND_PKT_BURST
		ofp_send_pending_pkt_burst();
#endif /*OFP_SEND_PKT_BURST*/
	}

	/* Never reached */
	return NULL;
}

static int webserver(void *arg)
{
	int serv_fd, tmp_fd;
	struct ofp_sockaddr_in my_addr;

	(void)arg;

	OFP_INFO("HTTP thread started");

	if (ofp_init_local()) {
		OFP_ERR("Error: OFP local init failed.\n");
		return -1;
	}
	sleep(1);

	myaddr = ofp_port_get_ipv4_addr(0, 0, OFP_PORTCONF_IP_TYPE_IP_ADDR);

	if ((serv_fd = ofp_socket(OFP_AF_INET, OFP_SOCK_STREAM, OFP_IPPROTO_TCP)) < 0) {
		OFP_ERR("ofp_socket failed");
		perror("serv socket");
		return -1;
	}

	memset(&my_addr, 0, sizeof(my_addr));
	my_addr.sin_family = OFP_AF_INET;
	my_addr.sin_port = odp_cpu_to_be_16(2048);
	my_addr.sin_addr.s_addr = myaddr;
	my_addr.sin_len = sizeof(my_addr);

	if (ofp_bind(serv_fd, (struct ofp_sockaddr *)&my_addr,
		       sizeof(struct ofp_sockaddr)) < 0) {
		OFP_ERR("Cannot bind http socket (%s)!", ofp_strerror(ofp_errno));
		return -1;
	}

	ofp_listen(serv_fd, 10);

#ifndef USE_EPOLL
	OFP_INFO("Using ofp_select");
	ofp_fd_set read_fd;
	OFP_FD_ZERO(&read_fd);
	int nfds = serv_fd;
#else
	OFP_INFO("Using ofp_epoll");
	int epfd = ofp_epoll_create(1);
	struct ofp_epoll_event e = { OFP_EPOLLIN, { .fd = serv_fd } };

static void send_arp_request(struct ofp_ifnet *dev, uint32_t gw)
{
	char buf[sizeof(struct ofp_ether_vlan_header) +
		sizeof(struct ofp_arphdr)];
	struct ofp_arphdr *arp;
	struct ofp_ether_header *e1 = (struct ofp_ether_header *)buf;
	struct ofp_ether_vlan_header *e2 =
					(struct ofp_ether_vlan_header *)buf;
	size_t size;
	odp_packet_t pkt;

	memset(buf, 0, sizeof(buf));
	memset(e1->ether_dhost, 0xff, OFP_ETHER_ADDR_LEN);
	memcpy(e1->ether_shost, dev->mac, OFP_ETHER_ADDR_LEN);

	if (ETH_WITH_VLAN(dev)) {
		arp = (struct ofp_arphdr *) (e2 + 1);
		e2->evl_encap_proto = odp_cpu_to_be_16(OFP_ETHERTYPE_VLAN);
		e2->evl_tag = odp_cpu_to_be_16(dev->vlan);
		e2->evl_proto = odp_cpu_to_be_16(OFP_ETHERTYPE_ARP);
		size = sizeof(*arp) + sizeof(*e2);
	} else {
		arp = (struct ofp_arphdr *) (e1 + 1);
		e1->ether_type = odp_cpu_to_be_16(OFP_ETHERTYPE_ARP);
		size = sizeof(*arp) + sizeof(*e1);
	}

	arp->hrd = odp_cpu_to_be_16(OFP_ARPHDR_ETHER);
	arp->pro = odp_cpu_to_be_16(OFP_ETHERTYPE_IP);
	arp->hln = OFP_ETHER_ADDR_LEN;
	arp->pln = sizeof(struct ofp_in_addr);
	arp->op  = odp_cpu_to_be_16(OFP_ARPOP_REQUEST);
	memcpy(arp->eth_src, e1->ether_shost, OFP_ETHER_ADDR_LEN);
	arp->ip_src = dev->ip_addr;
	memcpy(arp->eth_dst, e1->ether_dhost, OFP_ETHER_ADDR_LEN);
	arp->ip_dst = gw;

	pkt = ofp_packet_alloc(size);
	if (pkt == ODP_PACKET_INVALID) {
		OFP_ERR("ofp_packet_alloc failed");
		return;
	}

	memcpy(odp_packet_data(pkt), buf, size);

	if (odp_unlikely(ofp_if_type(dev) == OFP_IFT_VXLAN)) {
		ofp_vxlan_send_arp_request(pkt, dev);
		return;
	}

	odp_packet_has_eth_set(pkt, 1);
	odp_packet_has_arp_set(pkt, 1);
	odp_packet_l2_offset_set(pkt, 0);
	odp_packet_l3_offset_set(pkt, size - sizeof(struct ofp_arphdr));

	if (send_pkt_out(dev, pkt) == OFP_PKT_DROP)
		odp_packet_free(pkt);
}

int ofp_timer_cancel(odp_timer_t tim)
{
	odp_event_t timeout_event = ODP_EVENT_INVALID;
	odp_timeout_t tmo;
	uint32_t t = (uint32_t)tim;
	struct ofp_timer_internal *bufdata;
	struct ofp_timer_internal *prev = NULL;

	if (tim == ODP_TIMER_INVALID)
		return 0;

	if (t & 0x80000000) {
		/* long timeout */
		odp_spinlock_lock(&shm->lock);
		bufdata = shm->long_table[t & TIMER_LONG_MASK];

		while (bufdata) {
			struct ofp_timer_internal *next = bufdata->next;
			if (bufdata->id == t) {
				if (prev == NULL)
					shm->long_table[t & TIMER_LONG_MASK] = next;
				else
					prev->next = next;
				odp_buffer_free(bufdata->buf);
				odp_spinlock_unlock(&shm->lock);
				return 0;
			}
			prev = bufdata;
			bufdata = next;
		}
		odp_spinlock_unlock(&shm->lock);
		return -1;
	}
	else {
		if (odp_timer_cancel(tim, &timeout_event) < 0)
		{
			OFP_WARN("Timeout already expired or inactive");
			return 0;
		}

		if (timeout_event != ODP_EVENT_INVALID) {
			tmo = odp_timeout_from_event(timeout_event);
			bufdata = odp_timeout_user_ptr(tmo);
			odp_buffer_free(bufdata->buf);
			odp_timeout_free(tmo);
		} else {
			OFP_WARN("Lost timeout buffer at timer cancel");
			return -1;
		}

		if (odp_timer_free(tim) != ODP_EVENT_INVALID) {
			OFP_ERR("odp_timer_free failed");
			return -1;
		}
	}

	return 0;
}

int ofp_stat_lookup_shared_memory(void)
{
	shm_stat = ofp_shared_memory_lookup(SHM_NAME_STAT);
	if (shm_stat == NULL) {
		OFP_ERR("ofp_shared_memory_lookup failed");
		return -1;
	}
	return 0;
}

int ofp_uma_lookup_shared_memory(void)
{
	shm = ofp_shared_memory_lookup(SHM_NAME_UMA);
	if (shm == NULL) {
		OFP_ERR("ofp_shared_memory_lookup failed");
		return -1;
	}
	return 0;
}

static int ofp_uma_alloc_shared_memory(void)
{
	shm = ofp_shared_memory_alloc(SHM_NAME_UMA, sizeof(*shm));
	if (shm == NULL) {
		OFP_ERR("ofp_shared_memory_alloc failed");
		return -1;
	}
	return 0;
}

static int ofp_hook_alloc_shared_memory(void)
{
	shm_hook = ofp_shared_memory_alloc(SHM_NAME_HOOK, sizeof(*shm_hook));
	if (shm_hook == NULL) {
		OFP_ERR("ofp_shared_memory_alloc failed");
		return -1;
	}
	return 0;
}