 /*
  * (C) 2003-2004 Columbia University Departments of Computer Science and
  * Electrical Engineering
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, is permitted provided the original authors give
  * written consent and the following conditions are met:
  *
  * 1. Redistributions of source code must retain the above copyright
  *    notice, this list of conditions and the following disclaimer.
  * 2. Redistributions in binary form must reproduce the above
  *    copyright notice, this list of conditions and the following
  *    disclaimer in the documentation and/or other materials
  *    provided with the distribution.
  *
  * THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``AS IS'' AND
  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
  * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHORS
  * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
  * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
  * OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
  * TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
  * OF SUCH DAMAGE.
  */

void SendQueryData(struct Query q)
{
	struct sockaddr_in dst_addr;

	bzero((char *)&dst_addr, sizeof(dst_addr));
	dst_addr.sin_family = AF_INET;
	dst_addr.sin_addr.s_addr = q.To.ip.s_addr;
	dst_addr.sin_port = htons(q.To.port);

	sendto(chord_socket, (char *)&q, sizeof(q), 0, (struct sockaddr *)&dst_addr, sizeof(dst_addr));
	//printf("Sent Query of type = %d, to ip=%s, port=%d\n",q.QueryType & ~ResponseMask, inet_ntoa(q.To.ip), q.To.port);
}


void ForwardPacket(struct Query *q_response)
{

	struct Query newquery;
	UINT nextHop = GetNextHop(q_response->FinalDst.id);

	newquery = *q_response;
	newquery.From = GetMyLocation();
	newquery.To = GetLocation(nextHop);

	SendQueryData(newquery); // Forward the packet to the nextHop
}

void HandleResponse(struct Query *q_response)
{
	int index;
	if( !Stabilize_WakeUp(q_response) )
	{
		switch(q_response->QueryType & ~ResponseMask)
		{
		case QueryType_AskPredecessor:
//			printf("Response to AskPredecessor, but didn't wakeup any thread...\n");
//			printf("Probably had a timeout and now here I am.. unable to fo anything...\n");
			break;
		case QueryType_FindSuccessor:
			UpdateLocation(q_response->ResponseData);
			index = GetFingerIndex(q_response->QueryData.id);
			SetFinger(index, q_response->ResponseData.id);
			break;
		default:
			fprintf(stderr, "ERROR: Unknown QueryType found\n");
			exit(-3);
			break;
		}
	}
}


void HandleQuery(struct Query *query)
{
	UINT start, myID, mySucc;
	switch(query->QueryType)
	{
	case QueryType_AskPredecessor:
		struct Query q_response;
		q_response = MakeStandardQuery(query->OrigSrc.id, query->OrigSrc.id, QueryType_AskPredecessor | ResponseMask);
		q_response.QueryID = query->QueryID;
		q_response.QueryData = query->QueryData;
		q_response.ResponseData = GetLocation(GetMyPredecessor());
		SendQueryData(q_response);
		break;
	case QueryType_Notify:
		UpdateLocation(query->QueryData);
//		UpdatePredecessor(query->QueryData.id);
		SetPredecessor(query->QueryData.id);
		break;
	case QueryType_FindSuccessor:
		start = query->QueryData.id;
		myID = GetMyID();
		mySucc = GetMySuccessor();
		if(Belongs(start, GetIthStart(0), mySucc, 1, 1, FALSE))
		{
			struct Query q_response = MakeStandardQuery(query->OrigSrc.id, query->OrigSrc.id, QueryType_FindSuccessor | ResponseMask);
			q_response.QueryID = query->QueryID;
			q_response.QueryData = query->QueryData;
			q_response.ResponseData = GetLocation(mySucc);
			SendQueryData(q_response);
		}
		else
		{
			UINT nextNode = GetClosestPrecedingFinger(start);
			if(nextNode == myID)
				fprintf(stderr,"ERROR: Finger Table bad\n");
			struct Query q_forward = *query;
			q_forward.From = GetMyLocation();
			q_forward.To = q_forward.FinalDst = GetLocation(nextNode);
			SendQueryData(q_forward);
		}
		break;
	default:
		fprintf(stderr, "ERROR: Unknown QueryType found\n");
		exit(-2);
		break;
	}
}

void *RecvThread(void *arg)
{
	/*
	struct Thread_Input TI; // This is for passing the thread index and the query to the thread which'll
	// handle this query.. We assume that the thread will quickly copy this data into its local space,
	// so that this structure becomes redundant before we overwrite it at next packet recieve.
	*/

	char buf[MAXBUF];
	struct sockaddr *src_addr;
	socklen_t srclen;
	while(TRUE)
	{
		struct Query *q_response;
		int datalen = recvfrom(chord_socket, buf, MAXBUF, 0, src_addr, &srclen);
		if(datalen <= 0)
			continue;
//		printf("YO !!!!!!!!!!!!!!!!! Recieved A Packet of length %d\n",datalen);

		if(datalen < sizeof(struct Query))
		{
			printf("Error: Insufficient Data recieved\n");
			exit(-2);
		}
		// Process Data
		q_response = (struct Query *)buf;
//		Validate(q_response, src_addr);



//		printf("Updating From and OrigSrc in the HashTable\n");
		UpdateLocation(q_response->From);
		UpdateLocation(q_response->OrigSrc);
//`		printf("Finished Updating From and OrigSrc in the HashTable\n");

		
		// Now there can be 3 possible type of packets we can recieve
		// 1) Those for which FinalDst.id is not our ID.
		// 	In this case we just have to forward the packet to the appropriate nexthop
		// 2) A response to one of our earlier queries
		// 	In this case we store the query response and wake up the appropriate thread
		// 3) A query for us to respond
		// 	In this case we start a new thread to handle this query..


		// Case 1: FinalDst.id != myID
		if(q_response->FinalDst.id != GetMyID())
		{
//			printf("To be forwarded\n");
			ForwardPacket(q_response);
			/*
			int thread_index = FindAndLockThread();
			TI.index = thread_index;
			TI.query = *q_response;
			pthread_create(&Threads[thread_index].thread, NULL, ForwardPacket, &TI);
			WaitAMoment(); // Wait to give this thread a chance to copy TI to its local space..
			*/
			continue; // Get back to recieving packets
		}

		// Case 2: A Response to one of our earlier queries		
		if(q_response->QueryType & ResponseMask)
		{
//			printf("A Response to our query, type=%d\n",q_response->QueryType & ~ResponseMask);
			// A response to one of our queries...
			HandleResponse(q_response);
			/*
			int Qid = q_response->QueryID;
			int sem_index = CopyResponseData(Qid, q_response);
			if(sem_index >= 0)
				sem_post(&sems[sem_index].sem);
			*/
		}
		else
		{
			// A Query for us to respond !!!!
			printf("A Query for us to respond, type=%d\n",q_response->QueryType);
			HandleQuery(q_response);
			/*
			int thread_index = FindAndLockThread();
			printf("Found %d thread free\n",thread_index);
			TI.index = thread_index;
			TI.query = *q_response;
			pthread_create(&Threads[thread_index].thread, NULL, RespondToQuery, &TI);
			*/
		}
	}
}



struct Query MakeStandardQuery(UINT to, UINT finaldst, int QueryType)
{
	struct Query query;

	query.From = query.OrigSrc = GetMyLocation();
	query.To = GetLocation(to);
	query.FinalDst.id = finaldst;

	query.QueryID = GetNewQueryID();
	query.QueryType = QueryType;

	return query;
}


struct Query SendQueryAndGetResponse(struct Query query)
{
	struct Query q_response;
	Stabilize_SetupSleep(query.QueryID);
	// Let send the query
	SendQueryData(query);

	for(int time_slept=0;time_slept < MAX_POLL_TIME;time_slept += MIN_POLL_TIME)
	{
		usleep(MIN_POLL_TIME);
		if(Stabilize_CheckForResponse())
		{
			q_response = Stabilize_GetResponse();
			return q_response;
		}
	}
	q_response.ResponseData.id = UINT_ERROR;
	return q_response;
}

UINT Remote_AskPredecessor(UINT helper_id)
{
	printf("Sending AskPredecessor query to %d\n",helper_id);
	struct Query query, response;
	query = MakeStandardQuery(helper_id, helper_id, QueryType_AskPredecessor);
	query.QueryData = GetLocation(helper_id);

	response = SendQueryAndGetResponse(query);
	// Now we have the data in response
	struct location result = response.ResponseData;
	if(result.id != UINT_ERROR)
		UpdateLocation(result);
	return result.id;
}

void Remote_Notify(UINT successor)
{
	struct Query query, response;
	query = MakeStandardQuery(successor, successor, QueryType_Notify);
	query.QueryData = GetMyLocation();

	// Send this notification packet..
	SendQueryData(query);
}


void *SOS_Helper(void *arg)
{
	struct sockaddr_in my_addr;
	struct sockaddr remote_addr;
	socklen_t remote_len;
	char buffer[MAXBUF];

	bzero((char *)&my_addr, sizeof(my_addr));

// **************************
// D. Cook 1/03
// Start of changes for interface to route module
   int rte_mod_sock;
   struct location myloc = GetMyLocation();
   
   my_addr.sin_family = AF_INET;
   my_addr.sin_addr.s_addr = htonl(INADDR_ANY);
   my_addr.sin_port = htons(myloc.port);

   if ((sos_socket = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
       perror("communciation module socket error");
       exit(-4);
   }

   /* bind server port */
   if(bind(sos_socket, (struct sockaddr *) &my_addr, sizeof(my_addr))<0) {
      perror("cannot bind port ");
      exit(-5);
   }

   /* listen */
   if (listen(sos_socket,5) < 0) {
      perror("communication module listen error");
      exit(1);
   }

   /* accept connection */
   remote_len = sizeof(remote_addr);
   if ((rte_mod_sock = accept(sos_socket, (struct sockaddr *) &remote_addr, &remote_len)) < 0) {
      perror("communication module accept error ");
      exit(1);
   }
	
   while(TRUE) {
      int datalen = recv(rte_mod_sock, buffer, MAXBUF, 0);
      if(datalen <= 0) {perror("recvfrom error");}
      char result[MAXBUF];
      // Change by Sharat
      if(buffer)
      {
	      printf("Got string <%s> from Comm Module\n",buffer);
	      if(buffer[strlen(buffer) - 1] == '\n')
	      {
		      buffer[strlen(buffer) - 1] = 0;
	      }
	      printf("Changed string from Comm Module to <%s>\n",buffer);
         FindNextHop(buffer, result);
	 result[strlen(result) + 1] = 0;
	 result[strlen(result)] = '\n';
	     printf("Sending string to Comm Module to <%s>\n",result);
    //   strcpy(result, buffer); // for testing
      // printf("Got nextHop IP = %s\n",result);
        if (send(rte_mod_sock, result, strlen(result),0) < 0) {perror("sendto error");}
      }
   } // end of while
}
// ******** END OF CHANGES FOR ROUTE MODULE INTERFACE D.Cook 1/03


/*
UINT Remote_FindSuccessor(UINT helper_id, UINT id)
{
	printf("Sending Find_Successor query for %d to %d\n",id,helper_id);
	struct Query query, response;
	query = MakeStandardQuery(helper_id, helper_id, QueryType_FindSuccessor);
	query.QueryData.id = id;

	response = SendQueryAndGetResponse(query);
	// Now we have the data in response
	struct location result = response.ResponseData;
	UpdateLocation(result);
	return result.id;
}

UINT Remote_FindMySuccessor(UINT helper_id)
{
	return Remote_FindSuccessor(helper_id, GetMyID());
}

UINT Remote_FindPredecessor(UINT helper_id, UINT id)
{
	struct Query query, response;
	query = MakeStandardQuery(helper_id, helper_id, QueryType_FindPredecessor);
	query.QueryData.id = id;

	response = SendQueryAndGetResponse(query);
	// Now we have the data in response
	struct location result = response.ResponseData;
	UpdateLocation(result);
	return result.id;
}

UINT Remote_FindMyPredecessor(UINT helper_id)
{
	return Remote_FindPredecessor(helper_id, GetMyID());
}

void Remote_UpdateFingerTable(UINT dst, int k)
{
	struct Query query;
	query = MakeStandardQuery(dst, dst, QueryType_UpdateFingerTable);
	query.QueryData.id = k;

	SendQueryData(query);
}

UINT Remote_AskFingerEntry(UINT helper_id, int k)
{
	struct Query query, response;
	query = MakeStandardQuery(helper_id, helper_id, QueryType_AskFingerEntry);
	query.QueryData.id = k;

	response = SendQueryAndGetResponse(query);
	// Now we have the data in response
	struct location result = response.ResponseData;
	UpdateLocation(result);
	return result.id;
}


UINT Remote_FindMyFingerEntry(UINT helper_id, int k)
{
	if(k == 1)
		printf("\n\n\n@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@\n\n");
	return Remote_FindSuccessor(helper_id, GetIthStart(k));
}

void Remote_Notify(UINT successor)
{
	struct Query query, response;
	query = MakeStandardQuery(successor, successor, QueryType_Notify);
	query.QueryData = GetMyLocation();

	// Send this notification packet..
	SendQueryData(query);
}


struct Query SendQueryAndGetResponse(struct Query q)
{
	// Steps
	// 1) Find a semaphore which is free and lock it
	// 2) Send the query data through the global socket
	// 3) Wait on the semaphore
	// 4) Now copy the result from the query struct in your semaphore structure
	// 5) Unlock the semaphore
	// 6) Return the local query structure
	
	struct Query response;

	// Step 1:
	int sem_index = FindAndLockSemaphore(q.QueryID);
	if(sem_index < 0)
	{
		printf("Error: No. of threads exceeded MAX_THREADS\n");
		exit(-1);
	}

	// Step 2:
	SendQueryData(q);

	// Step 3:
	sem_wait(&sems[sem_index].sem);

	// Step 4:
	response = GetResponseData(sem_index);

	// Step 5:
	UnlockSemaphore(sem_index);

	// Step 6:
	return response;
}

void SendQueryData(struct Query q)
{
	struct sockaddr_in dst_addr;

	bzero((char *)&dst_addr, sizeof(dst_addr));
	dst_addr.sin_family = AF_INET;
//	dst_addr.sin_addr.s_addr = htonl(q.To.ip.s_addr);
	dst_addr.sin_addr.s_addr = q.To.ip.s_addr;
//	dst_addr.sin_addr.s_addr = htonl(INADDR_ANY);
//	inet_pton(AF_INET,"127.0.0.1",&dst_addr);
	dst_addr.sin_port = htons(q.To.port);

	sendto(sock_fd, (char *)&q, sizeof(q), 0, (struct sockaddr *)&dst_addr, sizeof(dst_addr));
	printf("Sent Query of type = %d, to ip=%s, port=%d\n",q.QueryType & ~ResponseMask, inet_ntoa(q.To.ip), q.To.port);
}


UINT Find_Predecessor(UINT id)
{
	if(id == GetMyID())
		return GetMyPredecessor();

	printf(" Outside id=%d, start=%d,end=%d\n",id,GetIthStart(0),GetMySuccessor());
	if(Belongs(id, GetIthStart(0), GetMySuccessor(), 1, 1))
	{
		printf("Inside id=%d, start=%d,end=%d\n",id,GetIthStart(0),GetMySuccessor());
		return GetMyID();
	}
	else
	{
		UINT nexthop = GetNextHop(id);
		printf("nexthop = %d\n",nexthop);
		return Remote_FindPredecessor(nexthop, id);
	}
}

*/
