#include <iostream>

#include <map>

#include <vector>

#include <string>

#include <string.h>

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#include <sys/wait.h>

#include <pthread.h>

#include <errno.h>

#include <sys/types.h>

#include <sys/socket.h>

#include <sys/stat.h>

#include <netinet/in.h>

#include <netdb.h>

#include <ctype.h>

#include <arpa/inet.h>

#include <signal.h>

#include <thread> // std::this_thread::sleep_for

#include <chrono>

#include <algorithm> // std::max

#include <fstream>

#include <deque> // std::deque

#include <list> // std::list

#include "queue.h"

#include <signal.h>

#include <dirent.h>

#include <time.h>

#define price_per_widget 10

#define TOTAL_NUM_WIDGETS 1000

#define TOTAL_AMT_MONEY 10000

#define EMPTY "EMPTY"

#define FILE_NAME "snapshot_"

#define GLOBAL_STATES "globalState_"

#define PORT00 3490

#define BACKLOG 10

#define MAXDATASIZE 256 // max number of bytes we can get at once

using namespace std;

typedef struct mp1_process_t_

}mp1_process;

typedef struct channel_state_t

}channel_state;

typedef struct receiver_t

}receiver;

vector<vector<string> > channels;

int send_message( int sender, unsigned int receiver, const char* msg);

void cast_marker(mp1_process* p)

}

unsigned int update_logic_timeStamp(unsigned int curr, unsigned int received)

}

void update_vector_timeStamp(vector<unsigned int >* curr, const vector<unsigned int >& received, int id )

}

void record_PState_to_file(const mp1_process* p)

}

void record_CState_to_file(mp1_process* p, unsigned int from, unsigned int logical, char* vtime, bool empty )

}

void change_process_state(mp1_process* p, long money, long widgets)

}

void init_channels( vector<vector<string> > & channels )

}

void sigchld_handler(int s)

{

while(waitpid(-1, NULL, WNOHANG) > 0);

}

void *get_in_addr(struct sockaddr *sa)

{

if (sa->sa_family == AF_INET) {

return &(((struct sockaddr_in*)sa)->sin_addr);

}

return &(((struct sockaddr_in6*)sa)->sin6_addr);

}

void get_vector_t(const char* vec, vector<unsigned int > & t )

}

void show_state(mp1_process* p)

}

void taking_snapshot(mp1_process* p) //MARKER: seq-1 from-1 logical-1 vector-1,1,1,1

}

void process_message(mp1_process* p, const char* msg )

}

static void* receiver_cb(void* arg)

{

receiver* worker=(receiver*)arg;

char* port_num=worker->listening_on_port;

int sockfd, new_fd, numbytes; // listen on sock_fd, new connection on new_fd

struct addrinfo hints, *servinfo, *p;

struct sockaddr_storage their_addr; // connector's address information

socklen_t sin_size;

struct sigaction sa;

int yes=1;

char s[INET6_ADDRSTRLEN];

int rv;

unsigned int parent_id=worker->parentID;

memset(&hints, 0, sizeof hints);

hints.ai_family = AF_UNSPEC;

hints.ai_socktype = SOCK_STREAM;

hints.ai_flags = AI_PASSIVE; // use my IP

if ((rv = getaddrinfo(NULL, port_num, &hints, &servinfo)) != 0) {

fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(rv));

return NULL;

}

// loop through all the results and bind to the first we can

for(p = servinfo; p != NULL; p = p->ai_next) {

if ((sockfd = socket(p->ai_family, p->ai_socktype,

p->ai_protocol)) == -1) {

perror("server: socket");

continue;

}

if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &yes,

sizeof(int)) == -1) {

perror("setsockopt");

exit(1);

}

if (bind(sockfd, p->ai_addr, p->ai_addrlen) == -1) {

close(sockfd);

perror("server: bind");

continue;

}

break;

}

if (p == NULL) {

fprintf(stderr, "server: failed to bind\n");

return NULL;

}

freeaddrinfo(servinfo); // all done with this structure

if (listen(sockfd, BACKLOG) == -1) {

perror("listen");

exit(1);

}

sa.sa_handler = sigchld_handler; // reap all dead processes

sigemptyset(&sa.sa_mask);

sa.sa_flags = SA_RESTART;

if (sigaction(SIGCHLD, &sa, NULL) == -1) {

perror("sigaction");

exit(1);

}

struct sockaddr_in* IPV4addr = (struct sockaddr_in * )(p->ai_addr) ;

unsigned int des_port = IPV4addr->sin_port ;

// printf("process#%u is waiting for connections from process#%u on port %d \n", parent_id, worker->from_process, des_port );

while(1) { // main accept() loop

char* mesg_stored=NULL;

sin_size = sizeof their_addr;

new_fd = accept(sockfd, (struct sockaddr *)&their_addr, &sin_size);

if (new_fd == -1)

{

perror("accept");

continue;

}

inet_ntop(their_addr.ss_family,

get_in_addr((struct sockaddr *)&their_addr),

s, sizeof s);

// printf("process#%u: channel C-%u%u established, and process#%u is ready to receive message from process#%u !\n",

// parent_id, worker->from_process, parent_id, parent_id, worker->from_process);

if ((numbytes = recv(new_fd, worker->message_received, MAXDATASIZE-1, 0)) == -1)

{

perror("recv");

exit(1);

}

worker->message_received[numbytes] = '\0';

asprintf(&mesg_stored, "%s", worker->message_received);

bzero(worker->message_received, MAXDATASIZE);

// if(parent_id==0)

// printf(" via C%d%d: process#%d received msg: %s , ready to enter critical section \n",

// worker->from_process , parent_id ,parent_id ,mesg_stored);

pthread_mutex_lock(&worker->parent->process_lock );

// if(parent_id==0)

// printf("process#%d with msg: %s is in critical section! Ready to enqueue the message \n", parent_id ,mesg_stored);

queue_enqueue(worker->parent->event_queue, (void*)mesg_stored );

pthread_cond_signal(&worker->parent->data_ready);

pthread_mutex_unlock(&worker->parent->process_lock );

mesg_stored=NULL;

close(new_fd); // parent doesn't need this

}

return NULL;

}

int send_message( int sender, unsigned int receiver, const char* buf)

}

void init_state(mp1_process* p)

}

void clear_process(mp1_process* p )

}

void test_connection_channels(mp1_process* p)

}

void sender(unsigned int id, int num, pid_t pid)

}

bool isNum(char* m)

}

void verify_condition(unsigned int num)

}

void search_snapshot(unsigned int num, unsigned int num_processes)

}

void parse_search_check(const char* cmd, int num_processes)