// funzione realtiva allo spinlock
void *do_something(long int who_i_am) {
    int i;
    int dummy;

    // imposta la cancellazione asincrona
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,&dummy);
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,&dummy);

    // aspetta che pure l'altro thread l'abbia fatto
    pthread_barrier_wait(&pbarrier);
    // sblocca il segnale SIGUSR1
    pthread_sigmask(SIG_UNBLOCK,&block_sig,NULL);

    while (1) {
        // selezione un intervallo casuale di tempo
        i=rand()%9+1;
        // avvisa
        printf("Thread %ld: going to sleep for %d sec...\n",who_i_am,i);
        // e dormi
        sleep(i);
        // prova ad acquisire lo spinlock
        if (pthread_spin_trylock(&spinlock)) {
            // in caso contrario avvisa
            printf("Thread %ld: wait for the spinlock...\n",who_i_am);
            // e attendi
            pthread_spin_lock(&spinlock);
        }
        // da qui ho acquisito lo spinlock
        // seleziona un intervallo casuale di tempo
        i=rand()%9+1;
        // avvisa
        printf("Thread %ld: spinlock obtained and now wait for %d sec...\n",who_i_am,i);
        // e dormi
        sleep(i);
        printf("Thread %ld: unlock spinlock...\n",who_i_am);
        // rilascia lo spinlock
        pthread_spin_unlock(&spinlock);
    }
    return NULL;
}

ssize_t
usdf_msg_recv(struct fid_ep *fep, void *buf, size_t len,
		void *desc, fi_addr_t src_addr, void *context)
{
	struct usdf_ep *ep;
	struct usdf_rx *rx;
	struct usdf_msg_qe *rqe;
	struct usdf_domain *udp;

	ep = ep_ftou(fep);
	rx = ep->ep_rx;
	udp = ep->ep_domain;

	if (TAILQ_EMPTY(&rx->r.msg.rx_free_rqe)) {
		return -FI_EAGAIN;
	}

	pthread_spin_lock(&udp->dom_progress_lock);

	rqe = TAILQ_FIRST(&rx->r.msg.rx_free_rqe);
	TAILQ_REMOVE(&rx->r.msg.rx_free_rqe, rqe, ms_link);
	--rx->r.msg.rx_num_free_rqe;

	rqe->ms_context = context;
	rqe->ms_iov[0].iov_base = buf;
	rqe->ms_iov[0].iov_len = len;
	rqe->ms_last_iov = 0;

	rqe->ms_cur_iov = 0;
	rqe->ms_cur_ptr = buf;
	rqe->ms_iov_resid = len;
	rqe->ms_length = 0;
	rqe->ms_resid = len;

	TAILQ_INSERT_TAIL(&rx->r.msg.rx_posted_rqe, rqe, ms_link);

	pthread_spin_unlock(&udp->dom_progress_lock);

	return 0;
}

inline int
sstack_sfsd_add(uint32_t weight, sstack_sfsd_pool_t *pools,
				sfsd_t *sfsd)
{
	sstack_sfsd_pool_t *temp = NULL;
	int i = 0;
	int index = -1;


	// Parameter validation
	if (weight < MAXIMUM_WEIGHT || weight > MINIMUM_WEIGHT ||
				NULL == pools || NULL == sfsd) {
		sfs_log(sfs_ctx, SFS_ERR, "%s: Invalid parametes specified \n",
						__FUNCTION__);

		errno = EINVAL;
		return -1;
	}

	// Get index of sfsd pool that covers the specified storeg weight
	for (i = 0; i < MAX_SFSD_POOLS; i++) {
		uint32_t low = *(uint32_t *) &weights[i][0];
		uint32_t high = *(uint32_t *) &weights[i][1];
		if (weight >= low && weight <= high)
			break;
	}
	index = i;
	temp = pools;

	for (i = 0; i < index; i++)
		temp ++;

	// Now temp points to the right pool
	pthread_spin_lock(&temp->lock);
	bds_list_add_tail((bds_list_head_t) &sfsd->list,
								(bds_list_head_t) &temp->list);
	pthread_spin_unlock(&temp->lock);

	return 0;
}

/**
	@brief This function set value to thread_specific data by the key.
*/
int pthread_setspecific (pthread_key_t key, const void *value)
{
    hurd_ihash_t speci_tb = (hurd_ihash_t)get_current_pt_specific();

    /* key is valid ? */
    if (key > __pthread_key_nums ||
            __destructort_arry[key] == DESTRUCTORT_INVALID)
        return -EINVAL;

    /* has created? */
    if (!speci_tb)
    {
        int ret;

        /* if failt, it must out of memory */
        if (hurd_ihash_create(&speci_tb, HURD_IHASH_NO_LOCP) != 0)
            return -ENOMEM;

        /* add speci_tb to speci_tables */
        pthread_spin_lock(&__pthread_specific_lock);
        ret = find_elem_index_and_add_to_arry(speci_tb, &__pthread_specific_arry, (int *)&__pthread_specific_nums, NULL);
        pthread_spin_unlock(&__pthread_specific_lock);

        /* errno is always no mem */
        if (ret < 0)
        {
            hurd_ihash_destroy(speci_tb);
            return -ENOMEM;
        }

        /* set to pthread */
        set_current_pt_specific((void *)speci_tb);
    }

    /* add to ihash tables */
    if (hurd_ihash_add(speci_tb, (hurd_ihash_key_t)key, (hurd_ihash_value_t)value) != 0)
        return -ENOMEM;

    return 0;
}

void calqueue_put(double timestamp, void *payload) {

	calqueue_node *new_node;

	//printf("calqueue: inserendo %f, cwidth %f, bukettop %f, nbuckets %d, lastprio %f\n", timestamp, cwidth, buckettop, nbuckets, lastprio);

	// Fill the node entry
	new_node = malloc(sizeof(calqueue_node));
	new_node->timestamp = timestamp;
	new_node->payload = payload;
	new_node->next = NULL;

	if(new_node == NULL){
		printf("Out of memory in %s:%d\n", __FILE__, __LINE__);
		abort();
	}

	pthread_spin_lock(&cal_spinlock);
	calqueue_enq(new_node);
	pthread_spin_unlock(&cal_spinlock);

}

static void
my_free_hook (void *ptr, const void *caller)
{
    pthread_spin_lock(&lock);
    /* Restore all old hooks */
    __malloc_hook = old_malloc_hook;
    __realloc_hook = old_realloc_hook;
    __free_hook = old_free_hook;
    /* Call recursively */
    free (ptr);
    /* Save underlying hooks */
    old_malloc_hook = __malloc_hook;
    old_realloc_hook = __realloc_hook;
    old_free_hook = __free_hook;
    /* printf might call free, so protect it too. */
    printf ("freed pointer %p\n", ptr);
    /* Restore our own hooks */
    __malloc_hook = my_malloc_hook;
    __realloc_hook = my_realloc_hook;
    __free_hook = my_free_hook;
    pthread_spin_unlock(&lock);
}

int pxp_init(void)
{
	pthread_spin_lock(&lock);
	if (fd > 0) {
		active_open_nr++;
		pthread_spin_unlock(&lock);
		return 0;
	}

	if (fd < 0) {
		fd = open(PXP_DEVICE_NAME, O_RDWR, 0);
		if (fd < 0) {
			pthread_spin_unlock(&lock);
			dbg(DBG_ERR, "open file error.\n");
			return -1;
		}
	}

	active_open_nr++;
	pthread_spin_unlock(&lock);
	return 0;
}

int nvmed_queue_complete(NVMED_QUEUE* nvmed_queue) {
	NVMED* nvmed;
	NVMED_IOD* iod;
	volatile struct nvme_completion *cqe;
	u16 head, phase;
	int num_proc = 0;
	nvmed = nvmed_queue->nvmed;
	
	pthread_spin_lock(&nvmed_queue->cq_lock);
	head = nvmed_queue->cq_head;
	phase = nvmed_queue->cq_phase;
	for(;;) {
		cqe = (volatile struct nvme_completion *)&nvmed_queue->cqes[head];
		if((cqe->status & 1) != nvmed_queue->cq_phase)
			break;

		if(++head == nvmed->dev_info->q_depth) {
			head = 0;
			phase = !phase;
		}
		
		iod = nvmed_queue->iod_arr + cqe->command_id;
		nvmed_complete_iod(iod);
		num_proc++;
		if(head == 0 || num_proc == COMPLETE_QUEUE_MAX_PROC) break;
	}
	if(head == nvmed_queue->cq_head && phase == nvmed_queue->cq_phase) {
		pthread_spin_unlock(&nvmed_queue->cq_lock);
		return num_proc;
	}

	COMPILER_BARRIER();
	*(volatile u32 *)nvmed_queue->cq_db = head;
	nvmed_queue->cq_head = head;
	nvmed_queue->cq_phase = phase;
	pthread_spin_unlock(&nvmed_queue->cq_lock);

	return num_proc;
}

static ssize_t
usdf_eq_write(struct fid_eq *feq, uint32_t event, const void *buf,
		size_t len, uint64_t flags)
{
	struct usdf_eq *eq;
	int ret;

	eq = eq_ftou(feq);

	pthread_spin_lock(&eq->eq_lock);

	/* EQ full? */
	if (atomic_get(&eq->eq_num_events) == eq->eq_ev_ring_size) {
		ret = -FI_EAGAIN;
		goto done;
	}

	ret = usdf_eq_write_event(eq, event, buf, len, flags);
done:
	pthread_spin_unlock(&eq->eq_lock);
	return ret;
}

MTC_STATIC  MTC_STATUS
script_terminate(void)
{
    MTC_U32 i;

    pthread_spin_lock(&lock);
    terminate = TRUE;
    for (i = 0 ; i < SCRIPT_SOCKET_NUM; i++) 
    {
        if (sc_listening_socket[i] > 0) 
        {
#if 0
            //  let exit system call handle this to give
            //  an accurate intication of daemon termination
            //  to calldaemon (command)

            close(sc_listening_socket[i]);
#endif
            sc_listening_socket[i] = -1;
        }
    }
    pthread_spin_unlock(&lock);

#if 0
    {
        int pthread_ret;
        // wait for thread termination;

        for (i = 0 ; i < SCRIPT_SOCKET_NUM; i++) 
        {
            if ((pthread_ret = pthread_join(sc_thread[i], NULL)) != 0) 
            {
                pthread_ret = pthread_kill(sc_thread[i], SIGKILL);
            }
        }
    }
#endif
    return MTC_SUCCESS;
}

//for tcp
//add to list, sentinel will delete this in the main event loop
int
delete_close_event(int fd, struct fetcher *f)
{
    struct list *el = NULL;
    struct list_node *nd = NULL;
    el = f->el;
    if (el == NULL)
        return -1;
    if ((nd = malloc(sizeof(struct list_node))) == NULL)
        return -1;
    nd->data = malloc(sizeof(int));
    if (nd->data == NULL) {
        free(nd);
        return -1;
    }
    memcpy(nd->data, &fd, sizeof(int));
    pthread_spin_lock(&el->lock);
    nd->next = el->head;
    el->head = nd;
    pthread_spin_unlock(&el->lock);
    return 0;
}

/*
 * Set this timer to fire "ms" milliseconds from now.  If the timer is already
 * queued, previous timeout will be discarded.
 *
 * When timer expires, the registered timer callback will be called and
 * the timer entry removed from the queued list.  The timer routine will not
 * be called again until usdf_timer_set() is called again to re-set it.
 * usdf_timer_set() is safe to call from timer service routine.
 */
int
usdf_timer_set(struct usdf_fabric *fp, struct usdf_timer_entry *entry,
		uint32_t ms)
{
	int ret;
	unsigned bucket;

	pthread_spin_lock(&fp->fab_timer_lock);

	/* If no timers active, cur_bucket_ms may need catchup */
	if (fp->fab_active_timer_count == 0) {
		fp->fab_cur_bucket_ms = usdf_get_ms();
		ret = usdf_fabric_wake_thread(fp);
		if (ret != 0) {
			goto out;
		}
	}

	if (entry->te_flags & USDF_TF_QUEUED) {
		LIST_REMOVE(entry, te_link);
		--fp->fab_active_timer_count;
	}

	// we could make "overflow" bucket...
	if (ms >= USDF_NUM_TIMER_BUCKETS) {
		ret = -FI_EINVAL;
		goto out;
	}
	bucket = (fp->fab_cur_bucket + ms) & (USDF_NUM_TIMER_BUCKETS - 1);

	LIST_INSERT_HEAD(&fp->fab_timer_buckets[bucket], entry, te_link);
	entry->te_flags |= USDF_TF_QUEUED;
	++fp->fab_active_timer_count;
	ret = 0;

out:
	pthread_spin_unlock(&fp->fab_timer_lock);
	return ret;
}

int mlx4_poll_cq(struct ibv_cq *ibcq, int ne, struct ibv_wc *wc)
{
	struct mlx4_cq *cq = to_mcq(ibcq);
	struct mlx4_qp *qp = NULL;
	int npolled;
	int err = CQ_OK;

	pthread_spin_lock(&cq->lock);

	for (npolled = 0; npolled < ne; ++npolled) {
		err = mlx4_poll_one(cq, &qp, wc + npolled);
		if (err != CQ_OK)
			break;
	}

	if (npolled)
		update_cons_index(cq);

	pthread_spin_unlock(&cq->lock);

	return err == CQ_POLL_ERR ? err : npolled;
}

static void *                   /* Loop 'arg' times incrementing 'glob' */
threadFunc(void *arg)
{
    int loops = *((int *) arg);
    int loc, j, s;

    for (j = 0; j < loops; j++) {
        s = pthread_spin_lock(&splock);
        if (s != 0)
            errExitEN(s, "pthread_spin_lock");

        loc = glob;
        loc++;
        glob = loc;

        s = pthread_spin_unlock(&splock);
        if (s != 0)
            errExitEN(s, "pthread_spin_unlock");
    }

    return NULL;
}

void tprintf(char *msg, ...)
{
	int ret;
	va_list vl;

	va_start(vl, msg);
	pthread_spin_lock(&buffer_lock);

	ret = vsnprintf(buffer + buffer_use,
			sizeof(buffer) - buffer_use,
			msg, vl);
	if (ret < 0)
		/* Something screwed up! Unexpected. */
		goto out;

	if (ret >= sizeof(buffer) - buffer_use) {
		tprintf_flush();

		/* Rewrite the buffer */
		ret = vsnprintf(buffer + buffer_use,
				sizeof(buffer) - buffer_use,
				msg, vl);
		/* Again, we've failed! This shouldn't happen! So 
		 * switch to vfprintf temporarily :-( */
		if (ret >= sizeof(buffer) - buffer_use) {
			fprintf(stderr, "BUG (buffer too large) -->\n");
			vfprintf(stdout, msg, vl);
			fprintf(stderr, " <--\n");
			goto out;
		}
	}

	if (ret < sizeof(buffer) - buffer_use)
		buffer_use += ret;

out:
	pthread_spin_unlock(&buffer_lock);
	va_end(vl);
}

void flush_buf(struct pipe* p, int id)
{
	struct pipe_elem* elem;

	// lock
	pthread_spin_lock(&p->lock);

	if (id < 0 || id >= p->n_dst)
		goto unlock;

	while (!dequeue(&p->dst[id], (void**)&elem)) {
		elem->ref_cnt--;
		if (elem->ref_cnt <= 0) {
			assert(elem->ref_cnt >= 0);
			assert(!enqueue(&p->src, elem));
		}
	}

unlock : 
	// unlock
	pthread_spin_unlock(&p->lock);
}

/*
 * I/O Completion of specific I/O
 * target_id : submission id
 */
void nvmed_io_polling(NVMED_HANDLE* nvmed_handle, u16 target_id) {
	NVMED* nvmed;
	NVMED_QUEUE* nvmed_queue;
	NVMED_IOD* iod;
	volatile struct nvme_completion *cqe;
	u16 head, phase;
	nvmed_queue = HtoQ(nvmed_handle);
	nvmed = HtoD(nvmed_handle);

	pthread_spin_lock(&nvmed_queue->cq_lock);
	while(1) {
		head = nvmed_queue->cq_head;
		phase = nvmed_queue->cq_phase;
		iod = nvmed_queue->iod_arr + target_id;
		if(iod->status == IO_COMPLETE) {
			break;
		}
		cqe = (volatile struct nvme_completion *)&nvmed_queue->cqes[head];
		for (;;) {
			if((cqe->status & 1) == nvmed_queue->cq_phase)
				break;
		}

		if(++head == nvmed->dev_info->q_depth) {
			head = 0;
			phase = !phase;
		}

		iod = nvmed_queue->iod_arr + cqe->command_id;
		nvmed_complete_iod(iod);

		COMPILER_BARRIER();
		*(volatile u32 *)nvmed_queue->cq_db = head;
		nvmed_queue->cq_head = head;
		nvmed_queue->cq_phase = phase;
	}
	pthread_spin_unlock(&nvmed_queue->cq_lock);
}

static void* fn_chld(void *arg)
{ 
	int rc = 0;

	/* Initialize spin lock */
	if(pthread_spin_init(&spinlock, PTHREAD_PROCESS_PRIVATE) != 0)
	{
		printf("main: Error at pthread_spin_init()\n");
		exit(PTS_UNRESOLVED);
	}

	/* Lock the spinlock */
	printf("thread: attempt spin lock\n");
	rc = pthread_spin_lock(&spinlock);
	if(rc != 0)
	{
		printf("Error: thread failed to get spin lock error code:%d\n" , rc);
		exit(PTS_UNRESOLVED);
	}
	printf("thread: acquired spin lock\n");
	
	/* Wait for main to try and unlock this spinlock */
	sem = INMAIN;
	while(sem == INMAIN)
		sleep(1);

	/* Cleanup just in case */
	pthread_spin_unlock(&spinlock);
	
	if(pthread_spin_destroy(&spinlock) != 0)
	{
		printf("Error at pthread_spin_destroy()");
		exit(PTS_UNRESOLVED);
	}	

	pthread_exit(0);
	return NULL;
}

int main(void)
{
	int rc = 0;

	printf("main: initialize spin lock\n");
	if (pthread_spin_init(&spinlock, PTHREAD_PROCESS_PRIVATE) != 0) {
		printf("main: Error at pthread_spin_init()\n");
		return PTS_UNRESOLVED;
	}

	printf("main: attempt spin lock\n");

	/* We should get the lock */
	if (pthread_spin_lock(&spinlock) != 0) {
		printf
		    ("Unresolved: main cannot get spin lock when no one owns the lock\n");
		return PTS_UNRESOLVED;
	}

	printf("main: acquired spin lock\n");

	printf("main: unlock spin lock\n");
	if (pthread_spin_unlock(&spinlock) != 0) {
		printf("main: Error at pthread_spin_unlock()\n");
		return PTS_UNRESOLVED;
	}

	printf("main: destroy spin lock\n");
	rc = pthread_spin_destroy(&spinlock);
	if (rc != 0) {
		printf("Test FAILED: Error at pthread_spin_destroy()"
		       "Return code : %d\n", rc);
		return PTS_FAIL;
	}

	printf("Test PASSED\n");
	return PTS_PASS;
}

void deallocate_vbufs(int hca_num)
{
    vbuf_region *r = vbuf_region_head;

#if !defined(CKPT)
    if (MPIDI_CH3I_RDMA_Process.has_srq
#if defined(RDMA_CM)
        || MPIDI_CH3I_RDMA_Process.use_rdma_cm_on_demand
#endif /* defined(RDMA_CM) */
        || MPIDI_CH3I_Process.cm_type == MPIDI_CH3I_CM_ON_DEMAND)
#endif /* !defined(CKPT) */
    {
        pthread_spin_lock(&vbuf_lock);
    }

    while (r)
    {
        if (r->mem_handle[hca_num] != NULL
            && ibv_dereg_mr(r->mem_handle[hca_num]))
        {
            ibv_error_abort(IBV_RETURN_ERR, "could not deregister MR");
        }

        DEBUG_PRINT("deregister vbufs\n");
        r = r->next;
    }

#if !defined(CKPT)
    if (MPIDI_CH3I_RDMA_Process.has_srq
#if defined(RDMA_CM)
        || MPIDI_CH3I_RDMA_Process.use_rdma_cm_on_demand
#endif /* defined(RDMA_CM) */
        || MPIDI_CH3I_Process.cm_type == MPIDI_CH3I_CM_ON_DEMAND)
#endif /* !defined(CKPT) */
    {
         pthread_spin_unlock(&vbuf_lock);
    }
}