task2.c

/*
# Copyright 2025 University of Kentucky
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# SPDX-License-Identifier: Apache-2.0
*/

/* 
Please specify the group members here

# Student #1: Charles Schmidt

*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <pthread.h>
#include <limits.h>
#include <math.h>

#define MAX_EVENTS 64
#define MESSAGE_SIZE 16
#define DEFAULT_CLIENT_THREADS 4
#define EPOLL_SAMPLING_RATE 100
#define TIMEOUT 1000
#define NUM_PIPE_PACKETS 5
#define SESSION_TABLE_TIMEOUT 3

char *server_ip = "127.0.0.1";
int server_port = 12345;
char *client_ip;
int client_port = 8080;
int num_client_threads = DEFAULT_CLIENT_THREADS;
int num_requests = 1000000;
socklen_t addr_len = sizeof(struct sockaddr_in);

typedef struct{
  int seq;
  int thread;
}packet_t;
int packet_size = sizeof(packet_t);
typedef struct {
    int epoll_fd;
    int socket_fd;
    int tx;
    int rx;
    int thread_num;
    long long total_rtt;
    long total_messages;
    float request_rate;
    struct sockaddr_in server_addr;
} client_thread_data_t;

int is_before(int ack_base,int seq_num){
  if(ack_base < NUM_PIPE_PACKETS-1){
    int num_wrap = NUM_PIPE_PACKETS-(ack_base+1);
    return(seq_num >=(NUM_PIPE_PACKETS*2)-(num_wrap+1) || seq_num<=ack_base);
  }
  if(ack_base >= NUM_PIPE_PACKETS+1){
    int num_wrap = ack_base-(NUM_PIPE_PACKETS);
    return (seq_num >= num_wrap && seq_num <= ack_base);
  }
  return seq_num <= ack_base;
}

void *client_thread_func(void *arg) {
    client_thread_data_t *data = (client_thread_data_t *)arg;
    struct epoll_event event, events[MAX_EVENTS];
    packet_t snd;
    packet_t rec;
    struct timeval send_timer, end, cur;
    struct timeval timer_array[NUM_PIPE_PACKETS];
    int send_base = 0;
    int ack_base  = -1;
    

    data->request_rate = 0;
    data->total_rtt = 0;
    data->total_messages = 0;

    // Register the socket in the epoll instance
    event.events = EPOLLIN;
    event.data.fd = data->socket_fd;
    epoll_ctl(data->epoll_fd, EPOLL_CTL_ADD, data->socket_fd, &event);
    gettimeofday(&send_timer,NULL);
    for(char i=0;i<NUM_PIPE_PACKETS;i++){
        snd.seq = i;
        snd.thread = data->thread_num;
        data->tx++;
        send(data->socket_fd, &snd, packet_size, 0);
        send_base = i;
        if(data->tx >= num_requests){
          break;
        }
    }

    while (1) {
        gettimeofday(&cur,NULL);
        long long micro = (cur.tv_sec - send_timer.tv_sec) * 1000000LL + (cur.tv_usec - send_timer.tv_usec);
        if(micro/1000 >= TIMEOUT&&data->tx<num_requests){
          gettimeofday(&send_timer,NULL);
          for(int i=ack_base+1;is_before(send_base,i%(NUM_PIPE_PACKETS*2));i++){
            snd.seq = i%(NUM_PIPE_PACKETS*2);
            snd.thread = data->thread_num;
            if(send(data->socket_fd, &snd, packet_size, 0)<0){
              perror("timeout sendto failure");
            }

          }
        }
        if((send_base == ack_base) && (micro/1000<TIMEOUT) && (data->tx<num_requests)){
          gettimeofday(&send_timer,NULL);
          for (int i=0;i<NUM_PIPE_PACKETS;i++){
            snd.seq = (send_base+1)%(NUM_PIPE_PACKETS*2);
            snd.thread = data->thread_num;
            if(send(data->socket_fd, &snd, packet_size, 0)<0){
              perror("timeout sendto failure");
            }
            send_base++;
            send_base = send_base%(NUM_PIPE_PACKETS*2);
            data->tx++;
          }
        }
        
        
                // use a low timeout for the epoll wait so that the function can keep checking the individual
        // packets timers like a sampling rate 
        int n_events = epoll_wait(data->epoll_fd, events, MAX_EVENTS, EPOLL_SAMPLING_RATE);
        for (int i = 0; i < n_events; i++) {
            if (events[i].data.fd == data->socket_fd) {
                
                int n = recv(data->socket_fd, &rec, packet_size,0);
                int seq_num = rec.seq;
                if(!is_before(ack_base, seq_num)){
                  ack_base = seq_num;
                  data->rx++;
                }else if(seq_num == ack_base){
                  data->rx++;
                }
                if(n<0){
                  perror("ack recieve failed");
                }
            }
        }
        gettimeofday(&end, NULL);
        long long rtt = (end.tv_sec - send_timer.tv_sec) * 1000000LL + (end.tv_usec - send_timer.tv_usec);
        if(data->tx >= num_requests && rtt/1000 >= TIMEOUT) break;
    }
    close(data->socket_fd);
    close(data->epoll_fd);
    return NULL;
}

void run_client() {
    pthread_t threads[num_client_threads];
    client_thread_data_t thread_data[num_client_threads];
    struct sockaddr_in server_addr,client_addr;

    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(server_port);
    if(inet_pton(AF_INET, server_ip, &server_addr.sin_addr)<0){
      perror("bad server ip");
    }

    // Create client threads
    for (int i = 0; i < num_client_threads; i++) {
        thread_data[i].epoll_fd = epoll_create1(0);
        thread_data[i].socket_fd = socket(AF_INET,SOCK_DGRAM,0);
        thread_data[i].server_addr = server_addr;
        thread_data[i].tx = 0;
        thread_data[i].rx = 0;
        thread_data[i].thread_num = i;
        if(connect(thread_data[i].socket_fd,(struct sockaddr*)&server_addr,addr_len)<0){
          perror("client socket failed to connect");
        }
    }

    for (int i = 0; i < num_client_threads; i++) {
        pthread_create(&threads[i], NULL, client_thread_func, &thread_data[i]);
    }

    // Wait for threads to complete
    long long total_rtt = 0;
    long total_messages = 0;
    float total_request_rate = 0;
    double tx = 0.0;
    double rx = 0.0;
    for (int i = 0; i < num_client_threads; i++) {
        pthread_join(threads[i], NULL);
        total_rtt += thread_data[i].total_rtt;
        total_messages += thread_data[i].total_messages;
        total_request_rate += thread_data[i].request_rate;
        tx += thread_data[i].tx;
        rx += thread_data[i].rx;
    }
    printf("tx: %i\n",(int)tx);
    printf("rx: %i\n",(int)rx);
    printf("percentage packets loss: %f\n",1-(rx/tx));
}
void reset_session_table(int *table){
  for(int i=0;i<num_client_threads;i++){
    table[i] = 0;
  }
}
void run_server() {
    int server_fd = socket(AF_INET, SOCK_DGRAM, 0);
    struct sockaddr_in server_addr,client_addr;
    int session_table[num_client_threads];
    struct timeval session_timer,cur;

    reset_session_table(session_table);

    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(server_port);

    if(bind(server_fd,(struct sockaddr *)&server_addr,addr_len)<0){
      perror("server port failed to bind");
    }
    
    int epoll_fd = epoll_create1(0);
    if(epoll_fd<0){
      perror("epoll create failed");
    }
    struct epoll_event event, events[MAX_EVENTS];

    event.events = EPOLLIN;
    event.data.fd = server_fd;
    if(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &event)<0){
      perror("epoll ctl failed");
    } 
    
    while (1) {
        gettimeofday(&cur,NULL);
        long long micro = (cur.tv_sec - session_timer.tv_sec) * 1000000LL + (cur.tv_usec - session_timer.tv_usec);

        if(micro/1000000>SESSION_TABLE_TIMEOUT){ 
          reset_session_table(session_table);
        }

        int n_events = epoll_wait(epoll_fd, events, MAX_EVENTS, TIMEOUT);
        if(n_events<0){
          perror("epoll wait failed");
        }

        for (int i = 0; i < n_events; i++) {
            if (events[i].data.fd == server_fd) {
                // Handle client data
                gettimeofday(&session_timer,NULL);
                packet_t buffer;
                int n = recvfrom(server_fd, &buffer, packet_size, 0,(struct sockaddr *)&client_addr,&addr_len);
                if(n == -1){
                  perror("packet error");
                }
                int thread_num = buffer.thread;
                int seq_num = buffer.seq;
                if(seq_num == session_table[thread_num]){
                  session_table[thread_num] += 1;
                  session_table[thread_num] = (session_table[thread_num])%(NUM_PIPE_PACKETS*2);
                  if(sendto(server_fd, &buffer, packet_size, 0,(struct sockaddr *)&client_addr,addr_len)<0){
                    perror("server send failed");
                  }
                } 
            }
        }
    }

    close(server_fd);
    close(epoll_fd);
}

int main(int argc, char *argv[]) {
    if (argc > 1 && strcmp(argv[1], "server") == 0) {
        if (argc > 2) server_ip = argv[2];
        if (argc > 3) server_port = atoi(argv[3]);
        if (argc > 4) num_client_threads = atoi(argv[4]);
        run_server();
    } else if (argc > 1 && strcmp(argv[1], "client") == 0) {
        if (argc > 2) server_ip = argv[2];
        if (argc > 3) server_port = atoi(argv[3]);
        if (argc > 4) num_client_threads = atoi(argv[4]);
        if (argc > 5) num_requests = atoi(argv[5]);
        //ran out of time so the num requests has to be a multiple of 5 to prevent buggy behavior
        if(num_requests%5 != 0){
          printf("num requests must be a multiple of 5\n");
          return 1;
        }
        run_client();
    } else {
        printf("Usage: %s <server|client> [server_ip server_port num_client_threads/max connections num_requests]\n", argv[0]);
    }

    return 0;
}