/*==============================================================================
 * - msgQ_init()
 *
 * - init MSG_QUE struct, and add it to msgQ_list
 */
MSG_QUE *msgQ_init (MSG_QUE *pNewMsgQ,
                    uint32   max_num, 
                    uint32   max_len)
{
    int cpsr_c;

    if (max_num == 0 || max_len == 0) {
        return NULL;
    }

    if (pNewMsgQ == NULL) {
        pNewMsgQ = malloc(sizeof(MSG_QUE));
    }

    if (pNewMsgQ != NULL) {
        pNewMsgQ->max_num = max_num;
        pNewMsgQ->max_len = max_len;
        pNewMsgQ->cur_num = 0;
        dlist_init(&pNewMsgQ->msg_list);
        dlist_init(&pNewMsgQ->wait_send_list);
        dlist_init(&pNewMsgQ->wait_recv_list);

        cpsr_c = CPU_LOCK();
        dlist_add(&_G_msgQ_list, (DL_NODE *)pNewMsgQ);
        CPU_UNLOCK(cpsr_c);
    }

    return pNewMsgQ;
}

struct os *
os_create(void)
{
	struct os *newOS;

	newOS = os_alloc(MAX_CPU);
	if (newOS == NULL)
		return NULL;

	rw_lock_init(&newOS->po_mutex);
	logical_mmap_init(newOS);
	newOS->po_state = PO_STATE_CREATED;
	newOS->po_lpid = atomic_add(&cur_lpid, 1);

	/* add to global OS hash */
	lock_acquire(&hype_mutex);
	newOS->os_hashentry.he_key = newOS->po_lpid;
	ht_insert(&os_hash, &newOS->os_hashentry);
	lock_release(&hype_mutex);

	lock_init(&newOS->po_events.oe_lock);
	dlist_init(&newOS->po_events.oe_list);
	dlist_init(&newOS->po_resources);

	return newOS;
}

DList* postorder(QuadTree* tree){

  Stack *s = malloc(sizeof(Stack));
  dlist_init(s, &free);

  Queue *q = malloc(sizeof(Queue));
  dlist_init(q, &free);
  push(s, tree->root);
  QuadTreeNode** current_data = malloc(sizeof(QuadTreeNode*));

  while(pop(s, (void**)current_data) != -1){
    if(quadtree_is_visited(*current_data)){
      enqueue(q, *current_data);
    }else{
      quadtree_mark(*current_data);
      push(s, *current_data);
      if(!quadtree_is_leaf(*current_data)){
	DList* children = (*current_data)->children_list;
	DListElmt* current_elem = dlist_head(children);

	while(current_elem != NULL){
	  push(s, current_elem->data);
	  current_elem = dlist_next(current_elem);

	}
      }
    }
  }
  return q;
}

struct map *create_map(struct point2D *top_left,struct point2D *bottom_right,callback_ enter_callback,callback_ leave_callback)
{
	//����block�����
	uint32_t length = abs(top_left->x - bottom_right->x);
	uint32_t width = abs(top_left->y - bottom_right->y);
	
	uint32_t x_count = length % BLOCK_LENGTH == 0 ? length/BLOCK_LENGTH : length/BLOCK_LENGTH + 1;
	uint32_t y_count = width % BLOCK_LENGTH == 0 ? width/BLOCK_LENGTH : width/BLOCK_LENGTH + 1;
	
	struct map *m = calloc(1,x_count*y_count*sizeof(struct map_block)+sizeof(struct map));
	m->top_left = *top_left;
	m->bottom_right = *bottom_right;
	m->x_count = x_count;
	m->y_count = y_count;
	uint32_t x,y;
	for(y = 0; y < y_count; ++y)
		for(x = 0; x < x_count; ++x)
		{
			struct map_block * b = get_block(m,y,x);
			dlist_init(&b->aoi_objs);
			b->x = x;
			b->y = y;
		}
	dlist_init(&m->super_aoi_objs);
	m->enter_callback = enter_callback;
	m->leave_callback = leave_callback;
	return m;
}

static int rxm_recv_queue_init(struct rxm_recv_queue *recv_queue, size_t size)
{
	recv_queue->recv_fs = rxm_recv_fs_create(size);
	if (!recv_queue->recv_fs)
		return -FI_ENOMEM;
	dlist_init(&recv_queue->recv_list);
	dlist_init(&recv_queue->unexp_msg_list);
	return 0;
}

/* Init tty related queues and pcbs
 */
void init_tty()
{
    int i;
    for(i = 0; i < NUM_TERMINALS; i++)
    {
        tty_write_queues[i] = dlist_init();
        tty_read_queues[i] = dlist_init();
        tty_writing_procs[i] = NULL;
        tty_reading_procs[i] = NULL;
    }
}

static void mrail_recv_queue_init(struct fi_provider *prov,
				  struct mrail_recv_queue *recv_queue,
				  dlist_func_t match_recv,
				  dlist_func_t match_unexp,
				  mrail_get_unexp_msg_entry_func get_unexp_msg_entry)
{
	recv_queue->prov = prov;
	dlist_init(&recv_queue->recv_list);
	dlist_init(&recv_queue->unexp_msg_list);
	recv_queue->match_recv = match_recv;
	recv_queue->match_unexp = match_unexp;
	recv_queue->get_unexp_msg_entry = get_unexp_msg_entry;
}

struct rxd_rx_entry *rxd_get_rx_entry(struct rxd_ep *ep)
{
	struct rxd_rx_entry *rx_entry;
	if (freestack_isempty(ep->rx_entry_fs))
		return NULL;

	rx_entry = freestack_pop(ep->rx_entry_fs);
	rx_entry->key = rx_entry - &ep->rx_entry_fs->buf[0];
	dlist_init(&rx_entry->entry);
	dlist_init(&rx_entry->wait_entry);
	dlist_insert_tail(&rx_entry->entry, &ep->rx_entry_list);
	return rx_entry;
}

/* Init basic process management, timer and a dummy kernel proc
 */
void init_processes() {
    next_pid = 0;
    log_info("Inside %s", __func__);
    timer_init();
    log_info("Init timer done");
    ready_queue = dlist_init();
    wait_queue = dlist_init();
    if(!ready_queue || !ready_queue) {
        log_err("Cannot init ready queue!");
    }
    log_info("Init queue done");
    init_idle_proc();
    log_info("Init idle done");
}

struct msg_que* new_msgque(uint32_t syn_size,item_destroyer destroyer)
{
	pthread_once(&g_msg_que_key_once,msg_que_once_routine);
	struct msg_que *que = calloc(1,sizeof(*que));
	pthread_key_create(&que->t_key,delete_per_thread_que);
    dlist_init(&que->blocks);
	que->mtx = mutex_create();
	que->refbase.destroyer = delete_msgque;
    llist_init(&que->share_que);
    dlist_init(&que->can_interrupt);
	que->syn_size = syn_size;
	que->destroy_function = destroyer;
	get_per_thread_que(que,MSGQ_NONE);
	return que;
}

/*
 * define methods needed for the GNI fabric provider
 */
static int gnix_fabric_open(struct fi_fabric_attr *attr,
			    struct fid_fabric **fabric,
			    void *context)
{
	struct gnix_fid_fabric *fab;

	if (strcmp(attr->name, gnix_fab_name)) {
		return -FI_ENODATA;
	}

	fab = calloc(1, sizeof(*fab));
	if (!fab) {
		return -FI_ENOMEM;
	}

	/*
	 * set defaults related to use of GNI datagrams
	 */
	fab->n_bnd_dgrams = gnix_def_gni_n_dgrams;
	fab->n_wc_dgrams = gnix_def_gni_n_wc_dgrams;
	fab->datagram_timeout = gnix_def_gni_datagram_timeouts;

	fab->fab_fid.fid.fclass = FI_CLASS_FABRIC;
	fab->fab_fid.fid.context = context;
	fab->fab_fid.fid.ops = &gnix_fab_fi_ops;
	fab->fab_fid.ops = &gnix_fab_ops;
	_gnix_ref_init(&fab->ref_cnt, 1, __fabric_destruct);
	dlist_init(&fab->domain_list);

	*fabric = &fab->fab_fid;

	return FI_SUCCESS;
}

/*
 * Initialize all available memory monitors
 */
void ofi_monitor_init(void)
{
	fastlock_init(&uffd_monitor->lock);
	dlist_init(&uffd_monitor->list);

	fi_param_define(NULL, "mr_cache_max_size", FI_PARAM_SIZE_T,
			"Defines the total number of bytes for all memory"
			" regions that may be tracked by the MR cache."
			" Setting this will reduce the amount of memory"
			" not actively in use that may be registered."
			" (default: 0 no limit is enforced)");
	fi_param_define(NULL, "mr_cache_max_count", FI_PARAM_SIZE_T,
			"Defines the total number of memory regions that"
			" may be store in the cache.  Setting this will"
			" reduce the number of registered regions, regardless"
			" of their size, stored in the cache.  Setting this"
			" to zero will disable MR caching.  (default: 1024)");
	fi_param_define(NULL, "mr_cache_merge_regions", FI_PARAM_BOOL,
			"If set to true, overlapping or adjacent memory"
			" regions will be combined into a single, larger"
			" region.  Merging regions can reduce the cache"
			" memory footprint, but can negatively impact"
			" performance in some situations.  (default: false)");

	fi_param_get_size_t(NULL, "mr_cache_max_size", &cache_params.max_size);
	fi_param_get_size_t(NULL, "mr_cache_max_count", &cache_params.max_cnt);
	fi_param_get_bool(NULL, "mr_cache_merge_regions",
			  &cache_params.merge_regions);

	if (!cache_params.max_size)
		cache_params.max_size = SIZE_MAX;
}

void _hashtable_rehash(HASHTABLE *t, size_t newsize)
{
	ARRAY table;
	DLIST *bucket;
	size_t i, j, hash;
	DLIST_ITER it, end;

	array_init(&table, sizeof(DLIST));
	array_resize(&table, newsize);
	for (i = 0; i != newsize; ++i) {
		dlist_init((DLIST*)array_at(&table, i), t->element_size);
	}
	
	j = array_size(&t->table);
	for (i = 0; i != j; ++i) {
		bucket = (DLIST*)array_at(&t->table, i);
		if (dlist_size(bucket)) {
			end = dlist_end(bucket);
			for (it = dlist_begin(bucket); it != end; it = dlist_next(it)) {
				hash = _hashtable_hash(t, dlist_at(it)) % newsize;
				dlist_push((DLIST*)array_at(&table, hash), dlist_at(it));
			}	
		}
	}
	array_destroy(&t->table);
	memcpy(&t->table, &table, sizeof(ARRAY));
}

/* init timer
 */
void timer_init(void) {
    timer.round_robin_quantumn = DEFAULT_QUANTUMN;
    timer.tick = timer.round_robin_quantumn;
    log_info("Init delay queue");
    delay_queue = dlist_init();
    log_info("Done init delay queue");
}

static void util_fabric_init(struct util_fabric *fabric, const char *name)
{
	atomic_initialize(&fabric->ref, 0);
	dlist_init(&fabric->domain_list);
	fastlock_init(&fabric->lock);
	fabric->name = name;
}

int util_buf_pool_create_attr(struct util_buf_attr *attr,
			      struct util_buf_pool **buf_pool)
{
	size_t entry_sz;
	ssize_t hp_size;

	(*buf_pool) = calloc(1, sizeof(**buf_pool));
	if (!*buf_pool)
		return -FI_ENOMEM;

	(*buf_pool)->attr = *attr;

	entry_sz = (attr->size + sizeof(struct util_buf_footer));
	(*buf_pool)->entry_sz = fi_get_aligned_sz(entry_sz, attr->alignment);

	hp_size = ofi_get_hugepage_size();

	if ((*buf_pool)->attr.chunk_cnt * (*buf_pool)->entry_sz < hp_size)
		(*buf_pool)->attr.is_mmap_region = 0;
	else
		(*buf_pool)->attr.is_mmap_region = 1;

	if (!(*buf_pool)->attr.indexing.ordered)
		slist_init(&(*buf_pool)->list.buffers);
	else
		dlist_init(&(*buf_pool)->list.regions);

	return FI_SUCCESS;
}

struct rxd_x_entry *rxd_rx_entry_init(struct rxd_ep *ep,
			const struct iovec *iov, size_t iov_count, uint64_t tag,
			uint64_t ignore, void *context, fi_addr_t addr,
			uint32_t op, uint32_t flags)
{
	struct rxd_x_entry *rx_entry;

	rx_entry = rxd_get_rx_entry(ep, op);
	if (!rx_entry) {
		FI_WARN(&rxd_prov, FI_LOG_EP_CTRL, "could not get rx entry\n");
		return NULL;
	}

	rx_entry->peer = addr;
	rx_entry->flags = flags;
	rx_entry->bytes_done = 0;
	rx_entry->offset = 0;
	rx_entry->next_seg_no = 0;
	rx_entry->iov_count = iov_count;
	rx_entry->op = op;
	rx_entry->ignore = ignore;

	memcpy(rx_entry->iov, iov, sizeof(*rx_entry->iov) * iov_count);

	rx_entry->cq_entry.op_context = context;
	rx_entry->cq_entry.len = ofi_total_iov_len(iov, iov_count);
	rx_entry->cq_entry.buf = iov[0].iov_base;
	rx_entry->cq_entry.tag = tag;

	rx_entry->cq_entry.flags = ofi_rx_cq_flags(op);
	dlist_init(&rx_entry->entry);

	return rx_entry;
}

int sock_poll_open(struct fid_domain *domain, struct fi_poll_attr *attr,
		   struct fid_poll **pollset)
{
	struct sock_domain *dom;
	struct sock_poll *poll;

	if (attr && sock_poll_verify_attr(attr))
		return -FI_EINVAL;

	dom = container_of(domain, struct sock_domain, dom_fid);
	poll = calloc(1, sizeof(*poll));
	if (!poll)
		return -FI_ENOMEM;
	
	dlist_init(&poll->fid_list);
	poll->poll_fid.fid.fclass = FI_CLASS_POLL;
	poll->poll_fid.fid.context = 0;
	poll->poll_fid.fid.ops = &sock_poll_fi_ops;
	poll->poll_fid.ops = &sock_poll_ops;
	poll->domain = dom;
	atomic_inc(&dom->ref);

	*pollset = &poll->poll_fid;
	return 0;
}

static int fi_ibv_add_rai(struct dlist_entry *verbs_devs, struct rdma_cm_id *id,
		struct rdma_addrinfo *rai)
{
	struct verbs_dev_info *dev;
	struct verbs_addr *addr;
	const char *dev_name;

	if (!(addr = malloc(sizeof(*addr))))
		return -FI_ENOMEM;

	addr->rai = rai;

	dev_name = ibv_get_device_name(id->verbs->device);
	dlist_foreach_container(verbs_devs, struct verbs_dev_info, dev, entry)
		if (!strcmp(dev_name, dev->name))
			goto add_rai;

	if (!(dev = malloc(sizeof(*dev))))
		goto err1;

	if (!(dev->name = strdup(dev_name)))
		goto err2;

	dlist_init(&dev->addrs);
	dlist_insert_tail(&dev->entry, verbs_devs);
add_rai:
	dlist_insert_tail(&addr->entry, &dev->addrs);
	return 0;
err2:
	free(dev);
err1:
	free(addr);
	return -FI_ENOMEM;
}

int tcpx_conn_mgr_init(struct tcpx_fabric *tcpx_fabric)
{
	int ret;

	dlist_init(&tcpx_fabric->poll_mgr.list);
	fastlock_init(&tcpx_fabric->poll_mgr.lock);
	ret = fd_signal_init(&tcpx_fabric->poll_mgr.signal);
	if (ret) {
		FI_WARN(&tcpx_prov, FI_LOG_FABRIC,"signal init failed\n");
		goto err;
	}

	tcpx_fabric->poll_mgr.run = 1;
	ret = pthread_create(&tcpx_fabric->conn_mgr_thread, 0,
			     tcpx_conn_mgr_thread, (void *) tcpx_fabric);
	if (ret) {
		FI_WARN(&tcpx_prov, FI_LOG_FABRIC,
			"Failed creating tcpx connection manager thread");

		goto err1;
	}
	return 0;
err1:
	fd_signal_free(&tcpx_fabric->poll_mgr.signal);
err:
	fastlock_destroy(&tcpx_fabric->poll_mgr.lock);
	return ret;
}