/*
*функция производителя: читает файл строка за строкой
*в каждой строке выделяет числа, суммирует их
*и кладет результат в общий буфер - queueSource
*/
static void *producer(void* arg)
{   pthread_mutex_lock(myMutex);
		//если очередь переполнена, то производитель ждет, чтобы потребитель снял оттуда элемент
    while(queue_size(queueSource) > 100)
			pthread_cond_wait(onTakeCondition, myMutex);
    //парсит строку    
		char *line = NULL; int temp;
		size_t len = 0;
		ssize_t read;
    while ((read = getline(&line, &len, f)) != -1) {
          char *token;
          char *rest = line;
          int sumIntoStr = 0;
          token = strtok(rest, " ");
					while(token != NULL){
							temp = atoi(token);
							sumIntoStr += temp;
							token = strtok(NULL, " ");
					}
					printf("in producer: %d\n", sumIntoStr);
          //кладём сумму строки в очередь
          queue_enqueue(queueSource, sumIntoStr);
					if (queue_size(queueSource) - 1 <= 0) //очередь была пустая, а теперь нет - об этом нужно оповестить потребителя
					pthread_cond_signal(onPutCondition);
    }
		pthread_mutex_unlock(myMutex);
}

/**
 * Calls appropriate callback functions for a given rule.
 *
 * @param rules Queue of rules.
 * @param notify_target Callback function for targets.
 * @param notify_dep Callback function for dependencies.
 * @param notify_cmd Callback function for commands.
 * @return Void.
 */
static void notifyRules
(
	queue_t *rules,
	void (*notify_target)(char *),
	void (*notify_dep)(char *,char *),
	void (*notify_cmd)(char *, char *)
)
{
	int ruleIdx, i;
	
	for(ruleIdx=0; ruleIdx < queue_size(rules); ruleIdx++){

		rule_t *curRule = queue_at(rules, ruleIdx);
		notify_target(curRule->target);

		for(i=0; i < queue_size(curRule->deps); i++){
			char *dep = queue_at(curRule->deps, i);
			notify_dep(curRule->target, dep);
		}
		
		for(i=0; i < queue_size(curRule->commands); i++){
			char *cmd = queue_at(curRule->commands, i);
			notify_cmd(curRule->target, cmd);
		}
	}
}

/* callback function to catch and handle ctrl+c
 *
 * @param - int signal
 * @return - all threads terminate, each closing their socket, and any memory left unfreed is now freed
 */
void handlecc(int sig)
{
	while(queue_size(clients))
	{
		int *temp=queue_dequeue(clients);
		shutdown(*temp, 2);
		free(temp);
	}
	while(queue_size(tids))
	{
		pthread_t *tid=queue_dequeue(tids);
		void **dummy;
		pthread_join(*tid, dummy);
	}
	queue_destroy(tids);
	queue_destroy(clients);
	free(tids);
	free(clients);
	pthread_mutex_destroy(lock);
	free(lock);


	exit(0);
	return;		//should never be reached
}

/*******************************************************************************
 * @author	: Rohan Jyoti
 * @name	: addToML
 * @param	: Membership List to add to, Payload to add
 * @return	: void
 * @purpose	: Add a payload if it is non-existant in the memebrship list
 ******************************************************************************/
void addToML(queue_t *incomingML, mPayload_t *incomingPayload)
{	
	if(queue_size(incomingML) == 0)
	{
		//time(&incomingPayload->hb_timestamp);
		incomingPayload->udp_socket = -1;
		queue_enqueue(incomingML, (void *)incomingPayload);
	}
	else
	{
		//Recall that it is the IP address combined with the port that is unique
		char *ip_addr = incomingPayload->IP_address;
		int tPort = incomingPayload->port;
		
		unsigned int i;
		for(i=0; i<queue_size(incomingML); i++)
		{
			mPayload_t *tempPayload = (mPayload_t *)queue_at(incomingML, i);
			if((strcmp(ip_addr, tempPayload->IP_address) == 0) && (tPort == tempPayload->port))
				return; //meaning payload already exists in ML
		}
		
		//reach here indicating that payload not in ML
		//time(&incomingPayload->hb_timestamp);
		incomingPayload->udp_socket = -1;
		queue_enqueue(incomingML, (void *)incomingPayload);
	}
}

int remove_item_test(void)
{
  clear_queue();
  insert_item(42);
  _assert(front() == 42);
  _assert(queue_size() == 1);

  _assert(remove_item() == 42);
  _assert(queue_size() == 0);
  _assert(is_empty() == true);

  insert_item(12);
  insert_item(13);
  insert_item(14);

  _assert(front() == 12);
  _assert(rear() == 14);
  _assert(queue_size() == 3);

  _assert(remove_item() == 12);
  _assert(front() == 13);
  _assert(rear() == 14);
  _assert(queue_size() == 2);

  _assert(remove_item() == 13);
  _assert(front() == 14);
  _assert(rear() == 14);
  _assert(queue_size() == 1);

  clear_queue();
  return 0;
}

int
queue_push(Queue * q, int value)
{
  int newhead;

  assert(q != NULL);
#ifdef QUEUE_STATS
  q->push_count++;
  q->push_total_size += queue_size(q);
#endif
  if (q->head == 0) {
    newhead = q->arraysize - 1;
  } else {
    newhead = q->head - 1;
  }
  if (newhead == q->tail) {
    return -1;  // queue is full
  }
  q->head = newhead;
  q->data[newhead] = value;
  q->elements++;

  assert(q->elements == queue_size(q));

  return 0;
}

void multi_job_check (struct queue *queue)
{

    zlist_t *newjobs;
    json_t *jobs;

    ok (queue_size (queue) == 0,
        "queue is initially empty");
    if (!(jobs = json_pack ("[{s:I s:i s:i s:f s:i},"
                             "{s:I s:i s:i s:f s:i}]",
                            "id", 1,
                            "priority", 10,
                            "userid", 42,
                            "t_submit", 1.0,
                            "flags", 0,
                            "id", 2,
                            "priority", 11,
                            "userid", 43,
                            "t_submit", 1.1,
                            "flags", 1)))
        BAIL_OUT ("json_pack() failed");

    newjobs = submit_enqueue_jobs (queue, jobs);
    ok (newjobs != NULL,
        "submit_enqueue_jobs works");
    ok (queue_size (queue) == 2,
        "queue contains 2 jobs");
    ok (zlist_size (newjobs) == 2,
        "newjobs contains 2 jobs");
    submit_enqueue_jobs_cleanup (queue, newjobs);
    ok (queue_size (queue) == 0,
        "submit_enqueue_jobs_cleanup removed queue entries");

    json_decref (jobs);
}

int main(void)
{
    Student s1 = {45, "hello"},
            s2 = {49, "world"}, * sptr;
    pQueue q = queue_new(sizeof(int));
    pQueue ps = queue_new(sizeof(Student));

    int value = 10, v = 12, *ip;
    queue_enqueue(q, &value);
    queue_enqueue(q, &v);
    ip = (int *)queue_front(q);
    printf("%d\n", *ip);
    queue_dequeue(q);
    ip = (int *)queue_front(q);
    printf("%d\n", *ip);
    queue_free(q);

    queue_enqueue(ps, &s1);
    queue_enqueue(ps, &s2);
    printf("queue size: %ld\n", queue_size(ps));
    sptr = (Student *)queue_front(ps);
    printf("no: %d, name: %s\n", sptr->no, sptr->name);
    queue_dequeue(ps);
    printf("queue size: %ld\n", queue_size(ps));
    sptr = (Student *)queue_front(ps);
    printf("no: %d, name: %s\n", sptr->no, sptr->name);
    queue_free(ps);
    exit(EXIT_SUCCESS);
}

int
queue_put(Queue * q, int value)
{
  int newtail;
#ifdef QUEUE_STATS
  int size;
#endif

  assert(q != NULL);
#ifdef QUEUE_STATS
  q->put_count++;
  size = queue_size(q);
  if (size > q->put_max_size) {
    assert(size == q->put_max_size + 1);
    q->put_max_size = size;
  }
  q->put_total_size += size;
#endif
  newtail = (q->tail + 1) % q->arraysize;
  if (newtail == q->head) {
    return -1;  // queue is full
  }
  q->data[q->tail] = value;
  q->tail = newtail;
  q->elements++;

  assert(q->elements == queue_size(q));

  return 0;
}

static ssize_t device_read(struct file * fp, char __user * up, size_t sz, loff_t * off)
{
size_t out_sz = 0;

if( MINOR( fp->f_dentry->d_inode->i_rdev ) > MINOR_COUNT )
	{
	printk( KERN_ALERT "Bad Minor number in inode\n" );
	return out_sz;
	}

switch( MINOR( fp->f_dentry->d_inode->i_rdev ) )
	{
	case MINOR_META:
		//
		break;

	case MINOR_RX:
		while( sz && queue_size( &rx_queue ) )
			{
			put_user( queue_pop( &rx_queue ), up++ );
			out_sz++;
			}
		break;

	case MINOR_TX:
		while( sz && queue_size( &tx_queue ) )
			{
			put_user( queue_pop( &tx_queue ), up++ );
			out_sz++;
			}
		break;
	}

return 0;
}

void _CONNECT(int sd){

  dem.procCounter++;

  printf("CONNECT (QUEUESIZE   :%d)\n",queue_size());
  printf("        (NUM OF PROCS:%d)\n",dem.procCounter);

  proc_data* sendProc;
  proc* p;

  sendProc = (proc_data*)malloc(sizeof(proc_data));
  p = get_proc(sd);

  if(queue_size() > 0 || dem.procCounter > MAXPROC){

    sendProc->REQUEST = CANNOTENTER;
    p->queued = 1;

  }else{

    sendProc->REQUEST = CONNECT;

    usleep(500000);//1.0[sec]

  }

  send(sd,sendProc,sizeof(proc_data),0);
}

void *consumer(void *data)
{
    struct wsqueue *queue = data;

    puts("start consume loop");
    while (cont || queue_size(&queue->squeue.queue)) {
        struct list_node *node;

        pthread_mutex_lock(&queue->squeue.mutex);
        while (cont && !queue_size(&queue->squeue.queue))
            pthread_cond_wait(&queue->cond, &queue->squeue.mutex);
        pthread_mutex_unlock(&queue->squeue.mutex);

        node = wsqueue_pop(queue);
        if (node) {
            struct payload *data = (struct payload *)node;
            printf("consumer got value %d\n", data->data);
            free(node);
        }
    }

    puts("finish consume loop");

    return NULL;
}

void imprime(){
	int area,i;
	aux = NULL;

	printf("%d-%d",tempo,(tempo+1));
	for(area = 0; area < qtd_Tprocess; area++){

		if (processes!= NULL){
			iterador = processes;
			for(i = 0; i < queue_size((queue_t *) processes); i++){
				if (iterador->id == area) aux = iterador;
				else iterador = iterador->next;
			}
		}


		if (ready!= NULL){
			iterador = ready;
			for(i = 0; i < queue_size((queue_t *) ready); i++){
				if (iterador->id == area) aux = iterador;
				else iterador = iterador->next;
			}
		}

		if (processo_corrente != NULL){
			if (processo_corrente->id == area){
				aux = processo_corrente;
				processo_corrente->tempo_executado_total++;
			}
		}

		if (finesh!= NULL){
			iterador = finesh;
			for(i = 0; i < queue_size((queue_t *) finesh); i++){
				if (iterador->id == area) aux = iterador;
				else iterador = iterador->next;
			}
		}

		if (aux != NULL){
			switch (aux->estado_atual){ // 0 - nova, 1 - pronta, 2 - rodando, 3 - terminada
				case 0:
					printf("\t");
					break;
				case 1:
					printf("\t--");
					tempo_medio_espera++;
					break;
				case 2:
					printf("\t##");
					break;
				case 3:
					printf("\t");
					break;
			}
		}
	}
	printf("\n");
}

int32_t init_pingpong_queue(struct pingpong_queue *ppq,char *name,int32_t (*action)(),queue_t *destq,queue_t *errorq)
{
    ppq->name = name;
    ppq->destqueue = destq;
    ppq->errorqueue = errorq;
    ppq->action = action;
    ppq->offset = 1;
    return(queue_size(&ppq->pingpong[0]) + queue_size(&ppq->pingpong[1]));  // init mutex side effect
}

/**
 * Determines if tgdb has commands it needs to run.
 *
 * \return
 * true if can issue directly to gdb. Otherwise false.
 */
bool Ctgdb::Has_command_to_run()
{
	if (tgdb_client_is_client_ready (tcc) && 
			((queue_size (gdb_input_queue) > 0) ||
			 (queue_size (oob_input_queue) > 0)))
		return true;

	return false;
}

void compute_to(sieve *s, int max)
{
	s->max = max;
	s->numbers = malloc(sizeof(struct linked_queue));
	s->prime = malloc(sizeof(struct linked_queue));

	int i;
	for (i = 2; i <= max; i++)
	{
		enqueue(s->numbers, i);
	}

	int p;
	queue *temp = malloc(sizeof( struct linked_queue));

	do {
		//obtain the next prime p by removing the first value in the queue of numbers.
		p = dequeue(s->numbers);
		//put p into the queue of primes.
		enqueue(s->prime, p);


		int size = queue_size(s->numbers);
		int x;
		//go through the queue of numbers, eliminating numbers divisible by p.
		for (x = 0; x < size; x++)
		{
			int t = dequeue(s->numbers);

			if (t % p != 0)
			{
				enqueue(temp, t);
			}
		}
		// Put stored prime values into the numbers queue.
		s->numbers = temp;

	} while (p < sqrt(max));

	// all remaining values in numbers queue are prime, so transfer them to primes queue
	int size = queue_size(s->numbers);
	int z;
	for (z = 0; z < size; z++)
	{
		enqueue(s->prime, dequeue(s->numbers));
	}

	// Returns numbers queue to original state.
	for (i = 2; i <= max; i++)
	{
		enqueue(s->numbers, i);
	}

	temp = NULL;
	free(temp);
}

int depsAreFiles(queue_t * dep){
	int i= 0;
	if(queue_size(dep) <= 0)
		return 0;
	for( ; i < queue_size(dep); i++){
		if(isRule(queue_at(dep, i)))
			return 0;
	}
	return 1;
}

void queue_check(){
    printf("Testing Queue Functions\n");
    Queue *q;
    q = queue_init(); assert_i(queue_size(q), 0, "Check Queue Size after init");

    queue_enqueue(q, 1, -1); assert_i(queue_size(q), 1, "Check Queue Size after Enqueue");
    queue_enqueue(q, 2, -2); assert_i(queue_size(q), 2, "Check Queue Size after Enqueue");
    queue_enqueue(q, 3, -3); assert_i(queue_size(q), 3, "Check Queue Size after Enqueue");

    int x, y;
    queue_dequeue(q, &x, &y);
        assert_p(x, y, 1, -1, "Check Dequeue");
        assert_i(queue_size(q), 2, "Check Size after Dequeue");
    queue_dequeue(q, &x, &y);
        assert_p(x, y, 2, -2, "Check Dequeue");
        assert_i(queue_size(q), 1, "Check Size after Dequeue");

    queue_enqueue(q, 6, -6); assert_i(queue_size(q), 2, "Check Queue Size after Enqueue");
    queue_dequeue(q, &x, &y);
        assert_p(x, y, 3, -3, "Check Dequeue");
        assert_i(queue_size(q), 1, "Check Size after Dequeue");
    queue_dequeue(q, &x, &y);
        assert_p(x, y, 6, -6, "Check Dequeue");
        assert_i(queue_size(q), 0, "Check Size after Dequeue");

    queue_free(q);
}

void *run_thread(void *args)
{
	arg_t *run_args = (arg_t *)args;
	
	while(1)
	{
		if (queue_size(run_args->ready_r) == 0 && 
			queue_size(run_args->rest_r) == 0)
		{
			return NULL;	
		}
		sem_wait(run_args->ready_sem);
		while(queue_size(run_args->ready_r) == 0)
		{
			if (queue_size(run_args->ready_r) == 0 && 
				queue_size(run_args->rest_r) == 0)
			{
				sem_post(run_args->ready_sem);
				return NULL;	
			}
			sem_wait(run_args->ready_sem);
		}
		rule_t *rule;
		pthread_mutex_lock(run_args->m);
		rule = queue_dequeue(run_args->ready_r);
		pthread_mutex_unlock(run_args->m);

		while(queue_size(rule->commands))
		{
			char *command = queue_dequeue(rule->commands);
			if (system(command) != 0)
			{
				exit(1);	
			}
			free(command);
		}

		char *temp = rule->target;
		while(queue_size(rule->deps))
		{
			char *dep = queue_dequeue(rule->deps);
			free(dep);
		}
		rule_destroy(rule);
		free(rule);
		
		pthread_mutex_lock(run_args->m);
		int pSize = queue_size(run_args->ready_r);
		queue_enqueue(run_args->complete_t, temp);
		process_queues(run_args->rest_r, run_args->ready_r, run_args->complete_t);
		int cSize = queue_size(run_args->ready_r);
		pthread_mutex_unlock(run_args->m);
		int i;
		for(i = 0; i < cSize - pSize; i++)
		{
			sem_post(run_args->ready_sem);
		}
		sem_post(run_args->ready_sem);
	}
}

void DLL_EXPORT action(queue_t * action_object){
    puts("IT'S OVER 50!");
int sum = 100;
while(sum>=50&&queue_size(action_object)>0){
    sum=0;
    queue_remove(action_object);
    for(int i = 0; i < queue_size(action_object); i++)
    sum+=queue_get(action_object,i);
}
printf("SUM AFTER REMOVING %i\n", sum);
}