void DiskMgr::ReadSeg(int fileid, int segid, int diskid)
{
	LOG_INFO_NOENDL("fake waiting thread distribution: ");
	for (int i = 0; i < m_diskNumber; ++i) {
		LOG_NOTIME_NOENDL(m_diskInfo[i].waitingThread << "\t");
	}
	LOG_NOTIME_NOENDL(endl);
	Pthread_mutex_lock(&m_diskInfo[diskid].mutex);
	++m_diskInfo[diskid].waitingThread;
	while (m_diskInfo[diskid].readingThread >= m_curReadThreadNum) {
		pthread_cond_wait(&m_diskInfo[diskid].cond, &m_diskInfo[diskid].mutex);
	}
	++m_diskInfo[diskid].readingThread;
	--m_diskInfo[diskid].waitingThread;
	Pthread_mutex_unlock(&m_diskInfo[diskid].mutex);
	
	//usleep(200000);
	double temp = m_blockSize * 1.0 / m_diskBand;
	LOG_INFO("server fake read, sleep " << temp << "s");
	usleep((int)(temp * 1000000));
	
	Pthread_mutex_lock(&m_diskInfo[diskid].mutex);
	--m_diskInfo[diskid].readingThread;
	pthread_cond_broadcast(&m_diskInfo[diskid].cond);
	Pthread_mutex_unlock(&m_diskInfo[diskid].mutex);
	Pthread_mutex_lock(&m_mutex);
	++m_totalRead;
	Pthread_mutex_unlock(&m_mutex);
}

void
cerebrod_speaker_data_initialize(void)
{
#if !WITH_CEREBROD_NO_THREADS
    Pthread_mutex_lock(&speaker_data_init_lock);
#endif /* !WITH_CEREBROD_NO_THREADS */
    if (speaker_data_init)
        goto out;

#if !WITH_CEREBROD_NO_THREADS
    /*
     * Must lock in this initialization routine, b/c the update thread
     * in a metric module may call the update state function.
     */
    Pthread_mutex_lock(&metric_list_lock);
#endif /* !WITH_CEREBROD_NO_THREADS */

    metric_list = List_create((ListDelF)_destroy_speaker_metric_info);

    if (_setup_metric_modules() < 0)
        CEREBRO_EXIT(("_setup_metric_modules"));

#if !WITH_CEREBROD_NO_THREADS
    Pthread_mutex_unlock(&metric_list_lock);
#endif /* !WITH_CEREBROD_NO_THREADS */

    speaker_data_init++;
out:
#if !WITH_CEREBROD_NO_THREADS
    Pthread_mutex_unlock(&speaker_data_init_lock);
#endif /* !WITH_CEREBROD_NO_THREADS */
    ;                             /* in case !WITH_CEREBRO_NO_THREADS */
}

/* barrier_wait: block if 'here' is less than 'count' */
int barrier_wait(barrier_t *b)
{
	int signo, err;
	pthread_t tid, **tidvp;
	
	Pthread_mutex_lock(&b->b_lock);
	b->b_here++;
				/* add tid to b->b_tids */
	for (tidvp = b->b_tids; *tidvp != NULL; tidvp++)
		;		/* find free slot */
	*tidvp = Malloc(sizeof(pthread_t *));
	**tidvp = pthread_self();
	
	if (b->b_here < b->b_cnt) {
		Pthread_mutex_unlock(&b->b_lock);
		err = sigwait(&mask, &signo);
		if (err != 0)
			err_exit(err, "sigwait error");
		if (signo == SIGUSR1)
			return 0;
		return 1;	/* unknown signal */
	}
				/* signal threads at barrier */
	for (tidvp = b->b_tids; *tidvp != NULL; tidvp++) {
		tid = **tidvp;
		if (tid != 0 && tid != pthread_self()) {
			err = pthread_kill(tid, SIGUSR1);
			if (err != 0)
				err_exit(err, "pthread_kill error");
		}
	}
	Pthread_mutex_unlock(&b->b_lock);
	return BARRIER_SERIAL_THREAD;
}

void *
consume(void *arg)
{
	for ( ; ; ) {
		Pthread_mutex_lock(&nready.mutex);
		if (nready.nconsumer > 0) {
			Pthread_mutex_unlock(&nready.mutex);
			continue;
		} else if (nready.nget >= nitems) {
			Pthread_mutex_unlock(&nready.mutex);
			return(NULL);	
		} else {
			nready.nconsumer++;
			while (nready.nready == 0)
				Pthread_cond_wait(&nready.cond, &nready.mutex);
				nready.nready--;
				nready.nget++;
				nready.nconsumer--;
				Pthread_mutex_unlock(&nready.mutex);

				// if (buff[i] != i)
				// 	printf("buff[%d] = %d\n", i, buff[i]);
			
				*((int *) arg) += 1;
			}
	}
	
	return(NULL);
}

/* @pid = medium priority process pid */
void high(pid_t pid)
{
	int status;
	set_rt_prio(0, prio_min+2, policy);

	/* Must come after raising the priority */
	Pthread_mutex_lock(statep->mutex);
		statep->high_started = 1;
	Pthread_mutex_unlock(statep->mutex);

	Pthread_mutex_lock(resource);
	Pthread_mutex_lock(statep->mutex);
		statep->high_owns_resource = 1;
		if (!statep->low_owns_resource || !statep->medium_started) {
			statep->inversion = 0;
		}
	Pthread_mutex_unlock(statep->mutex);
	Pthread_mutex_unlock(resource);
	kill(pid, SIGKILL);	/* kill the medium thread */
	waitpid(pid, &status, 0);

	Pthread_mutex_lock(statep->mutex);

	if (statep->inversion)
		printf("Successfully used priority inheritance to handle an inversion\n");
	else {
		printf("No inversion incurred\n");
	}
	Pthread_mutex_unlock(statep->mutex);
}

void *
Consumer(void *v)
{
	struct thread_data *data = (struct thread_data *)v;
	unsigned sum = 0;
	int n, item;
	int waits = 0;
	int l = maxbuflen - 1;
	
	for (n = 0; ; ++n) {
		Pthread_mutex_lock(&buf_mutex);
		while (buflen == 0) {
			if (eof) {
				Pthread_mutex_unlock(&buf_mutex);
				goto done;
			}
			++waits;
			Pthread_cond_wait(&empty_cond, &buf_mutex);
		}
		item = buf[--buflen];
		if (buflen == l)
			Pthread_cond_broadcast(&full_cond);
		Pthread_mutex_unlock(&buf_mutex);

		ConsumeItem(item, &sum);
	}

done:
	data->sum = sum;
	printf("Consumer %d exit, %d items received, %d waits\n", 
		data->no, n, waits);
	return data;
}

int threadpool_task_over(threadpool_t *pool, int block, int timeout) {
    int ret;

    Pthread_mutex_lock(&pool->mutex);
    if (pool->task_queue.len != 0) {
        if (!block) {
            Pthread_mutex_unlock(&pool->mutex);
            return -1;
        } else {
            struct timespec ts;
            struct timeval  tv;
            gettimeofday(&tv, NULL);
            ts.tv_sec = tv.tv_sec + timeout;
            ts.tv_nsec = tv.tv_usec * 1000;

            while (pool->task_queue.len != 0) {
                if (timeout == 0) {
                    Pthread_cond_wait(&pool->task_over_cond, &pool->mutex);
                } else {
                    ret = Pthread_cond_timedwait(&pool->task_over_cond, 
                        &pool->mutex, &ts);
                    if (ret == 0) {
                        Pthread_mutex_unlock(&pool->mutex);
                        return 0;
                    } else if (ret == ETIMEDOUT) {
                        Pthread_mutex_unlock(&pool->mutex);
                        return -1;
                    }
                }
            }
        }
    }
    Pthread_mutex_unlock(&pool->mutex);
    return 0;
}

int threadpool_add_task(threadpool_t *pool, 
        void (*func)(void*), void *arg, int priority) {
    int tosignal = 0;
    task_t *tq = (task_t *)calloc(1, sizeof(*tq));
    if (!tq) {
        return -1;
    }

    tq->func = func;
    tq->arg = arg;
    tq->priority = priority;

    Pthread_mutex_lock(&pool->mutex);
    if (pool->threads_idle == 0 && pool->threads_num < pool->threads_max) {
        threadpool_thread_create(pool);
        ++pool->threads_idle;
        ++pool->threads_num;
    }
    tosignal = (pool->task_queue.len == 0)  ? 1 : 0;
    if (heap_insert(&pool->task_queue, tq) != 0) {
        free(tq);
        Pthread_mutex_unlock(&pool->mutex);
        return -1;
    }
    Pthread_mutex_unlock(&pool->mutex);
    if (tosignal) {
        Pthread_cond_broadcast(&pool->cond);
    }
    return 0;
}

/*
 * _cerebrod_listener_initialize
 *
 * perform listener initialization
 */
static void
_cerebrod_listener_initialize(void)
{
  int i;

  Pthread_mutex_lock(&listener_init_lock);
  if (listener_init)
    goto out;

  Pthread_mutex_lock(&listener_fds_lock);
  for (i = 0; i < conf.listen_message_config_len; i++)
    {
      if ((listener_fds[i] = _listener_setup_socket(i)) < 0)
        CEREBROD_EXIT(("listener fd setup failed"));
    }
  Pthread_mutex_unlock(&listener_fds_lock);

  if (!(clusterlist_handle = clusterlist_module_load()))
    CEREBROD_EXIT(("clusterlist_module_load"));
  
  if ((found_clusterlist_module = clusterlist_module_found(clusterlist_handle)) < 0)
    CEREBROD_EXIT(("clusterlist_module_found"));

  if (found_clusterlist_module)
    {
      if (clusterlist_module_setup(clusterlist_handle) < 0)
        CEREBROD_EXIT(("clusterlist_module_setup"));

      if (conf.debug && conf.listen_debug)
        {
          fprintf(stderr, "**************************************\n");
          fprintf(stderr, "* Cerebro Clusterlist\n");
          fprintf(stderr, "* -----------------------\n");
          fprintf(stderr, "* Using Clusterlist: %s\n", 
                  clusterlist_module_name(clusterlist_handle));
          fprintf(stderr, "**************************************\n");
        }
    }

  cerebrod_listener_data_initialize();

  for (i = 0; i < conf.forward_message_config_len; i++)
    {
      /* if the forward destination is local to the machine, don't forward */
      if (conf.forward_message_config[i].ip_is_local)
        continue;
      if (_forwarding_setup(i) < 0)
        CEREBROD_EXIT(("forwarding setup failed"));
    }

  listener_init++;
  Pthread_cond_signal(&listener_init_cond);
 out:
  Pthread_mutex_unlock(&listener_init_lock);
}

static void* thread_loop(void *arg) {
    threadpool_t *pool = (threadpool_t*)arg;
    task_t *t = NULL;
    struct timespec ts;
    struct timeval  tv;
    int ret;
    int tosignal;

    while (!pool->exit) {
        Pthread_mutex_lock(&pool->mutex);
        gettimeofday(&tv, NULL);
        ts.tv_sec = tv.tv_sec + POOL_MAX_IDLE;
        ts.tv_nsec = tv.tv_usec * 1000;

        while (pool->task_queue.len == 0) {
            ret = Pthread_cond_timedwait(&pool->cond, &pool->mutex, &ts);
            if (ret == 0) {
                if (pool->exit) {
                    goto EXIT;
                }
                break;
            } else if (ret == ETIMEDOUT) {
                goto EXIT;
            }
        }

        --pool->threads_idle;
        t = heap_remove(&pool->task_queue, 0);
        tosignal = (pool->task_queue.len == 0) ? 1 : 0;
        Pthread_mutex_unlock(&pool->mutex);

        if (tosignal) {
            Pthread_cond_broadcast(&pool->task_over_cond);
        }

        if (t) {
            t->func(t->arg);
            free(t);
        }

        Pthread_mutex_lock(&pool->mutex);
        ++pool->threads_idle;
        Pthread_mutex_unlock(&pool->mutex);
    }

    Pthread_mutex_lock(&pool->mutex);
EXIT:
    --pool->threads_idle;
    tosignal = --pool->threads_num ? 0 : 1;
    Pthread_mutex_unlock(&pool->mutex);
    if (tosignal) {
        Pthread_cond_broadcast(&pool->exit_cond);
    }
    return NULL;
}

/* add_to_list takes a cell and adds that cell the
 * the cache list. It locks the cache for writing so
 * that other threads cannot modify it. */
void add_to_list(struct cache_cell *cell)
{
	/* locks the cache for writing */
	Pthread_rwlock_wrlock(&cache_lock);
	/* if there is enough space in the cache,
	 * no eviction is needed. */
	if (cache_size + cell->size <= MAX_CACHE_SIZE)
	{
		cell->next = head;
		if (head != NULL)
			head->previous = cell;
		head = cell;
		cache_size += cell->size;
		cell->last_use = cache_time;
		
		Pthread_mutex_lock(&time_mutex);
		cache_time++;
		Pthread_mutex_unlock(&time_mutex);
	}
	/* if there is not enough space in the cache,
	 * eviction is needed. */
	else
	{
		struct cache_cell *tmp_cell, *ptr;
		int tmp_last_use;
		
		/* remove elements from cache so that there is enough
		 * space in the cache. */
		while (!(cache_size + cell->size <= MAX_CACHE_SIZE))
		{
			tmp_last_use = cache_time + 1;
			for (ptr = head; ptr != NULL; ptr = ptr->next)
				if (ptr->last_use < tmp_last_use)
				{
					tmp_last_use = ptr->last_use;
					tmp_cell = ptr;
				}
			remove_from_list(tmp_cell);
		}
		
		/* add cell to cache */
		cell->next = head;
		if (head != NULL)
			head->previous = cell;
		head = cell;
		cache_size += cell->size;
		cell->last_use = cache_time;
		
		Pthread_mutex_lock(&time_mutex);
		cache_time++;
		Pthread_mutex_unlock(&time_mutex);
	}
	Pthread_rwlock_unlock(&cache_lock);
	return;
}

void *
Producer(void *v)
{
	struct producer *p = (struct producer *)v;
	struct consumer *c;
	struct packet *pa, *old;
	unsigned seed = p->seed;
	int n, k, cons;
	unsigned sum = 0;
	int waits = 0;
	int more = 1;
	int item_cnt = 0;

	cons = seed % num_consumers;
	for (n = 0; more; ++n) {

		/* Hole Packet aus lokaler emptyPacketQueue */
		Pthread_mutex_lock(&p->queue_mutex);
		while (!p->emptyPacketQueue) {
			++waits;
			Pthread_cond_wait(&p->empty_cond, &p->queue_mutex);
		}
		pa = p->emptyPacketQueue;
		p->emptyPacketQueue = pa->next;
		Pthread_mutex_unlock(&p->queue_mutex);

		/* Fuelle Packet */
		for (k = 0; k < buflen; ++k) {
			pa->buf[k] = ProduceItem(&seed, &sum);
			if (++item_cnt == num_items) {
				more = 0;
				++k;
				break;
			}
		}
		pa->len = k;

		/* Versende Packet an Consumer cons */
		c = consumers + cons;
		Pthread_mutex_lock(&c->queue_mutex);
		old = pa->next = c->fullPacketQueue;
		c->fullPacketQueue = pa;
		Pthread_mutex_unlock(&c->queue_mutex);
		if (!old)
			Pthread_cond_broadcast(&c->empty_cond);

		if (++cons == num_consumers)
			cons = 0;
	}

	p->sum = sum;
	printf("Producer %d exit, %d waits, %d packets\n", p->no, waits, n);
	return NULL;
}

void consume_wait(int i)
{
	for(;;){
		Pthread_mutex_lock(&shared.mutex);
		if(i < shared.nput){
			Pthread_mutex_unlock(&shared.mutex);
			return ;
		}
		Pthread_mutex_unlock(&shared.mutex);
	}
}

void *
thread_main_dns(void *arg)
{
	int dnsrequest(const char *name, uint32 *ipaddr);
	int connfd, *number, count, total;
	char  *domain ;
	uint32 *ipptr;
	pthread_mutex_t *mutex;
	printf("dns thread %d starting \n", (int)arg );

	for(;;){

		Pthread_mutex_lock(&dns_array_mutex);

		while(iget == iput)
			Pthread_cond_wait(&dns_array_cond,
					&dns_array_mutex);
		
		printf("thread %d  working \n", pthread_self());

		connfd	= dns_array[iget].sockfd;
		domain  = dns_array[iget].domain;
		number	= dns_array[iget].number;
 		count	= dns_array[iget].count;
		total	= dns_array[iget].total;
		ipptr	= dns_array[iget].ipptr;
		mutex	= dns_array[iget].mutex;
		
		if(++iget == ARRAYSIZE)
			iget = 0;		

		Pthread_mutex_unlock(&dns_array_mutex);

		//do our job 
		if(dnsrequest(domain, ipptr + count) != 0) 
			inet_pton(AF_INET, "127.0.0.1", ipptr+count);
		dc_free(domain);	//dc_free the memory of domain
		Pthread_mutex_lock(mutex);
		(*number)--;
		Pthread_mutex_unlock(mutex);
		if((*number) == 0){
			write(connfd, ipptr, total*sizeof(uint32));
			printf("sended\n");
	//		close(connfd);	//todo
			dc_free(ipptr);
			dc_free(number);
			dc_free(mutex);
		} 


	}

}

/* miss_handler handles miss case. It passes client's request
 * to server and passes server's response to client. If the 
 * response satisfies size requirements, store the response 
 * into cache. */
void miss_handler(int clientfd, struct client_request *request)
{
	int serverfd, length, response_size;
	char buf[MAXBUF], object_buf[MAX_OBJECT_SIZE];
	struct cache_cell *cell;
	rio_t rio_server;
	
	response_size = length = 0;
	
	/* acts as a client and writes request to server */
	Pthread_mutex_lock(&open_mutex);
	serverfd = open_serverfd_h(request, clientfd);
	Pthread_mutex_unlock(&open_mutex);
	rio_readinitb(&rio_server, serverfd);
	
	if (rio_writen(serverfd, request->request, request->request_length) 
					!= request->request_length)
	{
		write(2, "write error\n", strlen("write error\n"));
		close_connection(request, clientfd, serverfd);
	}
	
	/* passes server's response to client */
	while (1)
	{
		if ((length = rio_readnb(&rio_server, buf, MAXBUF)) < 0)
			close_connection(request, clientfd, serverfd);
		if (response_size + length <= MAX_OBJECT_SIZE)
			memcpy(object_buf + response_size, buf, length);
		response_size += length;
		if (rio_writen(clientfd, buf, length) < length)
			break;
		if (length != MAXBUF)
			break;
	}
	
	/* if response satisfies size requirement, store the response 
	 * into cache */
	if (response_size <= MAX_OBJECT_SIZE)
	{
		/* need a mutex to prevent inserting the same cell twice 
		 * into cache in race condition */
		Pthread_mutex_lock(&dup_mutex);
		if (search_cell_variant(request->request_line) == 0)
		{
			cell = allocate_cell();
			set_cell(cell, request->request_line, object_buf, response_size);
			add_to_list(cell);
		}
		Pthread_mutex_unlock(&dup_mutex);
	}
	close_connection(request, clientfd, serverfd);
}

/* include consume */
void
consume_wait(int i)
{
	for ( ; ; ) {
		Pthread_mutex_lock(&shared.mutex);
		if (i < shared.nput) {
			Pthread_mutex_unlock(&shared.mutex);
			return;			/* an item is ready */
		}
		Pthread_mutex_unlock(&shared.mutex);
		sched_yield();
	}
}

/* 
 * cerebrod_monitor_modules_update
 *
 * Send metric data to the appropriate monitor modules, if necessary.
 * The struct cerebrod_node_data lock should already be locked.
 */
void
cerebrod_monitor_modules_update(const char *nodename,
                                struct cerebrod_node_data *nd,
                                const char *metric_name,
                                struct cerebrod_message_metric *mm)
{
  struct cerebrod_monitor_module_info *monitor_module;
  struct cerebrod_monitor_module_list *ml;
#if CEREBRO_DEBUG
  int rv;
#endif /* CEREBRO_DEBUG */
  
  assert(nodename && nd && metric_name && mm);

  if (!monitor_index)
    return;

#if CEREBRO_DEBUG
  /* Should be called with lock already set */
  rv = Pthread_mutex_trylock(&nd->node_data_lock);
  if (rv != EBUSY)
    CEREBROD_EXIT(("mutex not locked: rv=%d", rv));
#endif /* CEREBRO_DEBUG */

  if ((ml = Hash_find(monitor_index, metric_name)))
    {
      ListIterator itr = NULL;

      Pthread_mutex_lock(&(ml->monitor_list_lock));
      itr = List_iterator_create(ml->monitor_list);
      Pthread_mutex_unlock(&(ml->monitor_list_lock));

      while ((monitor_module = list_next(itr)))
	{
	  Pthread_mutex_lock(&monitor_module->monitor_lock);
	  monitor_module_metric_update(monitor_handle,
				       monitor_module->index,
				       nodename,
				       metric_name,
				       mm->metric_value_type,
				       mm->metric_value_len,
				       mm->metric_value);
	  Pthread_mutex_unlock(&monitor_module->monitor_lock);
	}

      Pthread_mutex_lock(&(ml->monitor_list_lock));
      List_iterator_destroy(itr);
      Pthread_mutex_unlock(&(ml->monitor_list_lock));
    }
}

/* free_cell frees a cell and destroys the lock in that cell */
void free_cell(struct cache_cell *cell)
{
	if (cell->content != NULL)
	{
		Pthread_mutex_lock(&free_mutex);
		free(cell->content);
		Pthread_mutex_unlock(&free_mutex);
	}
	Pthread_rwlock_destroy(&(cell->lock));
	Pthread_mutex_lock(&free_mutex);
	free(cell);
	Pthread_mutex_unlock(&free_mutex);
	return;
}

int threadpool_destroy(threadpool_t *pool, int block, int timeout) {
    int ret;
    assert(pool);
    Pthread_mutex_lock(&pool->mutex);
    if (!pool->exit) {
        /* you should call `threadpool_exit' first */
        Pthread_mutex_unlock(&pool->mutex);
        return -1;
    }

    if (pool->threads_num != 0) {
        if (!block) {
            Pthread_mutex_unlock(&pool->mutex);
            return -1;
        } else {
            struct timespec ts;
            struct timeval  tv;
            gettimeofday(&tv, NULL);
            ts.tv_sec = tv.tv_sec + timeout;
            ts.tv_nsec = tv.tv_usec * 1000;

            while (pool->threads_num != 0) {
                if (timeout == 0) {
                    Pthread_cond_wait(&pool->exit_cond, &pool->mutex);
                    goto CONT;
                } else {
                    ret = Pthread_cond_timedwait(&pool->exit_cond, 
                        &pool->mutex, &ts);
                    if (ret == 0) {
                        goto CONT;
                    } else if (ret == ETIMEDOUT) {
                        Pthread_mutex_unlock(&pool->mutex);
                        return -1;
                    }
                }
            }
        }
    }
 
CONT:
    Pthread_mutex_unlock(&pool->mutex);
    heap_destroy(&pool->task_queue);
    Pthread_mutex_destroy(&pool->mutex);
    Pthread_cond_destroy(&pool->cond);
    Pthread_cond_destroy(&pool->exit_cond);
    Pthread_cond_destroy(&pool->task_over_cond);
    free(pool);
    return 0;
}

/* @pid = high priority process pid */
void low(pid_t pid)
{
	int status;
	Pthread_mutex_lock(resource);
		Pthread_mutex_lock(statep->mutex);
			statep->low_owns_resource = 1;
			if (statep->high_owns_resource ||
					statep->medium_started) {
				statep->inversion = 0;
			}
		Pthread_mutex_unlock(statep->mutex);
		usleep(500);
	Pthread_mutex_unlock(resource);
	waitpid(pid, &status, 0);
}