Networking

CMakeLists.txt

@@ -3,7 +3,7 @@ cmake_minimum_required(VERSION 3.13)
 # set the project name
 project(OlinCoin C CXX)
 enable_testing()
-#set(CMAKE_BUILD_TYPE Debug)
+# set(CMAKE_BUILD_TYPE Debug)
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 set(CMAKE_BINARY_DIR ${CMAKE_BINARY_DIR})
 set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})

src/core/blocks/create_block.c

@@ -23,7 +23,6 @@ unsigned long calc_block_reward(unsigned long blockchain_height){
   return BLOCK_REWARD;
 }
 
-
 Transaction *create_coinbase_tx(unsigned long tx_fees){
   Transaction *coinbase_tx = malloc(sizeof(Transaction));
   Output *miner_output = malloc(sizeof(Output));

src/core/txs/validate_tx.c

@@ -1,12 +1,12 @@
 /**
  * @file validate_tx.c
  * @author Nathan Faber nfaber@olin.edu
- * @brief 
+ * @brief
  * @version 0.1
  * @date 2022-03-20
- * 
+ *
  * @copyright Copyright (c) 2022
- * 
+ *
  */
 #include "base_tx.h"
 #include "utxo_pool.h"
@@ -39,7 +39,7 @@ void create_blank_sig_txhash(unsigned char *blank_hash, Transaction *tx){
     memset(blank_sig_tx->inputs[i].signature, 0, SIGNATURE_LEN);
   }
   memcpy(blank_sig_tx->outputs, tx->outputs, tx->num_outputs*sizeof(Output));
-  
+
   // need to make hash where the tx has signatures that are 0's
   hash_tx(blank_hash, blank_sig_tx);
   free_tx(blank_sig_tx);
@@ -111,9 +111,9 @@ int validate_coinbase_tx_parts_not_null(Transaction *coinbase_tx){
   if(coinbase_tx->num_inputs != 0){
     return 2;
   }
-  if(coinbase_tx->inputs != NULL){
-    return 3;
-  }
+  /* if(coinbase_tx->inputs != NULL){ */
+  /*   return 3; */
+  /* } */
   if(coinbase_tx->num_outputs != 1){
     return 4;
   }
@@ -122,7 +122,7 @@ int validate_coinbase_tx_parts_not_null(Transaction *coinbase_tx){
   }
   return 0;
 }
-  
+
 int validate_tx_shared(Transaction *tx){
   unsigned long total_in = 0;
   unsigned long total_out = 0;
@@ -192,4 +192,3 @@ int validate_tx_incoming(Transaction *tx){
   }
   return 0;
 }
-

src/core/utils/init_db.c

@@ -13,7 +13,7 @@ int create_folder(char *path){
   int mkdir_res = mkdir(path, 0777);
   if(mkdir_res != 0 && errno != EEXIST){
     //errors here
-    return 1;   
+    return 1;
   }
   return 0;
 }
@@ -23,7 +23,7 @@ int create_proj_folders(){
   if(!home_path){
     fprintf(stderr, "$HOME env variable not read\n");
     exit(1);
-  }  
+  }
   char *newPath = malloc(strlen(home_path) + strlen(LOCAL_LOCATION) + 1);
   strcpy(newPath, home_path);
   strcat(newPath, LOCAL_LOCATION);
@@ -78,7 +78,7 @@ int init_db(leveldb_t **db, char **dest, char *db_env, char *name){
   if(!home_path){
     fprintf(stderr, "$HOME env variable not read\n");
     exit(1);
-  }  
+  }
   *dest = malloc(strlen(home_path) + strlen(LOCAL_LOCATION) + strlen(db_env) + strlen(name) + 1);
   strcpy(*dest, home_path);
   strcat(*dest, LOCAL_LOCATION);
@@ -99,7 +99,7 @@ int destroy_db(leveldb_t **db, char *name){
   char *err = NULL;
   int ret = 0;
   options = leveldb_options_create();
- 
+
   /* DESTROY */
   leveldb_destroy_db(options, name, &err);
   if (err != NULL) {
@@ -126,4 +126,4 @@ int db_count(leveldb_t *db, unsigned int *num_entries){
   leveldb_readoptions_destroy(roptions);
   *num_entries = count;
   return 0;
-}
+}

src/includes/utils/queue.h

@@ -3,8 +3,6 @@
 #include <stdlib.h>
 #include <pthread.h>
 #include <semaphore.h>
-#include "base_tx.h"
-#include "base_block.h"
 
 typedef struct QueueItem{
   void *item;

src/runtime/CMakeLists.txt

@@ -15,6 +15,25 @@ target_link_libraries(shell
   handle_block
   init_globals)
 
+add_library(server server.c)
+target_compile_options(server PRIVATE ${COMPILE_OPTIONS_STD})
+target_include_directories(server PUBLIC
+  ${INCLUDE_ALL}
+  )
+target_link_libraries(server
+  core_block
+  rt
+)
+
+add_library(client client.c)
+target_compile_options(client PRIVATE ${COMPILE_OPTIONS_STD})
+target_include_directories(client PUBLIC
+  ${INCLUDE_ALL}
+  )
+target_link_libraries(client
+  core_block
+  rt
+)
 add_executable(runtime runtime.c)
 target_compile_options(runtime PRIVATE ${COMPILE_OPTIONS_STD})
 target_include_directories(runtime PUBLIC ${INCLUDE_ALL})
@@ -25,4 +44,16 @@ target_link_libraries(runtime
   init_globals
   validate_tx
   handle_tx
-  shell)
+  shell
+  server
+  client)
+
+add_executable(q_server queue_server.c)
+target_compile_options(q_server PRIVATE ${COMPILE_OPTIONS_STD})
+target_include_directories(q_server PUBLIC ${INCLUDE_ALL})
+target_link_libraries(q_server rt)
+
+add_executable(q_client queue_client.c)
+target_compile_options(q_client PRIVATE ${COMPILE_OPTIONS_STD})
+target_include_directories(q_client PUBLIC ${INCLUDE_ALL})
+target_link_libraries(q_client rt)

src/runtime/client.c

@@ -0,0 +1,204 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <errno.h>
+#include <string.h>
+#include <netdb.h>
+#include <sys/types.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+
+#include <arpa/inet.h>
+#include "client.h"
+#include "server.h"
+#include "runtime.h"
+#include "ser_block.h"
+#include "ser_tx.h"
+
+char peers[][20] = {"192.168.32.251"};//"ubuntu.local"};//"localhost",
+unsigned int num_peers = 1; // Match above
+
+void *get_in_addr2(struct sockaddr *sa)
+{
+	if (sa->sa_family == AF_INET) {
+		return &(((struct sockaddr_in*)sa)->sin_addr);
+	}
+
+	return &(((struct sockaddr_in6*)sa)->sin6_addr);
+}
+
+int recv_all(int s, char *buf, size_t *len) {
+  size_t total = 0;        // how many bytes we've read
+  size_t bytes_left = *len; // how many we have left to read
+  int n;
+
+  while (total < *len) {
+    n = recv(s, buf + total, bytes_left, 0);
+    if (n == 0)
+      return 1;
+    else if (n == -1)
+      return -1;
+    total += n;
+    bytes_left -= n;
+  }
+
+  *len = total; // return number actually sent here
+
+  return 0;
+}
+
+int recv_obj(int s, char *buf, size_t *total_size) {
+  int rv;
+  size_t head_size, buf_size;
+
+  head_size = sizeof(int) + sizeof(long);
+  if ((rv = recv_all(s, buf, &head_size)) != 0)
+    return rv;
+
+  buf_size = *(size_t*)(buf + sizeof(int));
+  if ((rv = recv_all(s, buf + head_size, &buf_size)) != 0)
+    return rv;
+
+  *total_size = buf_size + head_size;
+  return 0;
+}
+
+void client_fork(Globals *globals, pid_t pid, int i) {
+  int sockfd, rv, priority;
+  size_t numbytes;
+  unsigned long counter;
+  char buf[MAX_MSG_SIZE];
+  struct sockaddr_in serv_addr;
+  struct mq_attr attr;
+  mqd_t incoming;
+
+  if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
+      printf("\n Socket creation error \n");
+      exit(0);
+  }
+
+  serv_addr.sin_family = AF_INET;
+  serv_addr.sin_port = htons(PORT_INT);
+
+  // Convert IPv4 and IPv6 addresses from text to binary form
+  if (inet_pton(AF_INET, peers[i], &serv_addr.sin_addr) <= 0) {
+      printf("\nInvalid address/ Address not supported \n");
+      exit(0);
+  }
+  if (connect(sockfd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {
+      printf("\nConnection Failed \n");
+      exit(0);
+  }
+
+  attr.mq_flags = 0;
+  attr.mq_maxmsg = MAX_MESSAGES;
+  attr.mq_msgsize = MAX_MSG_SIZE;
+  attr.mq_curmsgs = 0;
+  incoming = mq_open(
+    globals->q_client,
+    O_WRONLY | O_CREAT,
+    QUEUE_PERMISSIONS,
+    &attr
+  );
+  if (incoming == -1) {
+    perror ("Client: mq_open (server)");
+    exit (1);
+  }
+  printf("Opened Incoming Queue for writing\n");
+
+  counter = 0;
+  while (pid != 1) {
+    // Check to see if parent killed
+    if (counter % 10000) {
+      pid = getpid();
+    }
+
+    // Get data over the socket
+    rv = recv_obj(sockfd, buf, &numbytes);
+    if (rv == -1) {
+      perror("recv obj");
+      exit(1);
+    } else if (rv == 1) {
+      // Connection has been closed from the other side (by the socket server)
+      break;
+    }
+
+    // Add the data to the queue
+    priority = *(int*)buf;
+    if (mq_send(incoming, buf, numbytes, priority) == -1) {
+        perror ("Client: Not able to send message to main process");
+        continue;
+    }
+    printf("client: received %lu bytes\n", numbytes);
+
+    counter++;
+  }
+
+  close(sockfd);
+  exit(0);
+}
+
+void *client_thread(void *arg){
+  Globals *globals = arg;
+  for (unsigned int i = 0; i < num_peers; i++) {
+    // Fork this process and connect!
+    pid_t pid = fork();
+    if (!pid) { // this is the child process
+      client_fork(globals, pid, i);
+		}
+  }
+
+  //INCOMING PARENT
+  struct mq_attr attr;
+  mqd_t incoming_parent;
+  char in_buffer[MSG_BUFFER_SIZE], *ser_buffer;
+  unsigned int priority;
+  int id;
+  size_t buf_size;
+
+  attr.mq_flags = 0;
+  attr.mq_maxmsg = MAX_MESSAGES;
+  attr.mq_msgsize = MAX_MSG_SIZE;
+  attr.mq_curmsgs = 0;
+  incoming_parent = mq_open(
+    globals->q_client,
+    O_RDONLY | O_CREAT,
+    QUEUE_PERMISSIONS,
+    &attr
+  );
+  if (incoming_parent == -1) {
+    perror ("Client: mq_open (server)");
+    exit (1);
+  }
+
+  // Now pop off the queues whenever something is added!
+  while (1) {
+    if (mq_receive(incoming_parent, in_buffer, MAX_MSG_SIZE, &priority) == -1) {
+      perror ("Client: mq_receive");
+      exit (1);
+    }
+
+    id = *(int*)in_buffer;
+    buf_size = *(size_t*)(in_buffer + sizeof(int));
+    ser_buffer = in_buffer + sizeof(int) + sizeof(size_t);
+
+    printf(
+      "Client parent recieved %lu byte object of id '%d'\n",
+      buf_size, id
+    );
+    if (id == BLOCK_ID) {
+      Block *new_block = deser_block_alloc(NULL, (unsigned char *)ser_buffer);
+      printf("Block received over network:\n");
+      print_block(new_block, "");
+      queue_add_void(globals->queue_block, new_block);
+    }
+    else if (id == TX_ID) {
+      Transaction *new_tx = deser_tx_alloc(NULL, (unsigned char *)ser_buffer);
+      printf("Tx received over network:\n");
+      print_tx(new_tx, "");
+      queue_add_void(globals->queue_tx, new_tx);
+    }
+  }
+
+  return 0;
+}

src/runtime/includes/client.h

@@ -0,0 +1,4 @@
+#define MAXDATASIZE 100 // max number of bytes we can get at once 
+
+
+void *client_thread();