Merge pull request #3 from gistrec/develop

Delete multithreading, remove unnecessary functions

Author: gistrec

src/Config.hpp

@@ -6,7 +6,7 @@
 std::string fileName;
 int mtu; // Max packet size to send and receive
 
-std::atomic<int> ttl; // Max time to wait new packets
+int ttl; // Max time to wait new packets
 int ttl_max;
 
 SOCKET _socket;

src/Main.cpp

@@ -64,7 +64,7 @@ int main(int argc, char* argv[]) {
             std::cerr << "Cant't get access to broadcast" << std::endl;     //
             exit(1);                                                        //
         }                                                                   // If option 'broadcast' is yes
-        broadcast_address.sin_addr.s_addr = INADDR_ANY;                     // change server address
+        broadcast_address.sin_addr.s_addr = INADDR_BROADCAST;               // change server address
     } else {                                                                // to broadcast
         broadcast_address.sin_addr.s_addr =                                 // Else change server address
             inet_addr(result["broadcast"].as<std::string>().c_str());       // to address in option
@@ -77,6 +77,16 @@ int main(int argc, char* argv[]) {
         exit(1);                                                                  //
     }                                                                             //
 
+    #if defined(_WIN32) || defined(_WIN64)                                      //
+    int tv = 1 * 1000;  // user timeout in milliseconds [ms]                    //
+    setsockopt(_socket, SOL_SOCKET, SO_RCVTIMEO, (char*)&tv, sizeof(tv));       // Set socket
+    #else                                                                       // receive timeout
+    struct timeval tv;                                                          //        to 1 sec
+    tv.tv_sec = 1;                                                              //
+    tv.tv_usec = 0;                                                             //
+    setsockopt(_socket, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof(tv)); //
+    #endif
+
     // Run receiver or sender
     if (result["type"].as<std::string>() == "receiver") {        //
         Receiver::run(result);                                   //
@@ -87,4 +97,4 @@ int main(int argc, char* argv[]) {
     }
 
     return 0;
-}
+}

src/Receiver.cpp

@@ -22,11 +22,10 @@ std::vector<int> getEmptyParts() {
 }
 
 /**
- * Run in secondary thread
  * Start while receive packet 'FINISH'
  * Get empty part's and requests it's from the server
  */
-void async checkParts() {
+void checkParts() {
     char* buffer = new char[100];
 
     std::vector<int> emptyParts = getEmptyParts();
@@ -40,29 +39,36 @@ void async checkParts() {
 
             std::cout << "Requested " << index << " part" << std::endl;
         }
-        std::this_thread::sleep_for(1s);
         emptyParts = getEmptyParts();
     }
 
     std::ofstream output(fileName, std::ofstream::binary);      //
     output.write(file, file_length);                            // Save file
-    system(std::string("CertUtil -hashfile " + fileName + " SHA256").c_str());             //
 }
 
 
 void run(cxxopts::ParseResult &options) {
-    std::thread(checkTTL); // Create thread, to check server timeout
+    bool finish = false; // Sender finish transfering
 
-    char* buffer = new char[2 * mtu];
+    char* buffer;
 
     while (auto length = recvfrom(_socket, buffer, 2 * mtu, 0, (sockaddr*) &server_address, &server_address_length)) {
+        // Sender is no longer available
+        if (ttl <= 0) return;
+
+        // If Sender finish transfering check missing parts every 1 sec
+        if (finish) checkParts();
+
+        if (length <= 0) continue; // If no more packets in buffer
+
         ttl = ttl_max; // Update ttl
 
         if (strncmp(buffer, "NEW_PACKET", 10) == 0) {
             file_length = Utils::getIntFromBytes(buffer + 10, 4);
+            mtu = Utils::getIntFromBytes(buffer + 14, 4);
 
+            buffer = new char[2 * mtu];
             file = new char[file_length];
-            memset(file, 0, file_length);
 
             std::cout << "Receive information about new file size: " << file_length << std::endl;
             std::cout << "Part count: " << int((float)file_length / (float)mtu + 0.5) << std::endl;
@@ -74,10 +80,11 @@ void run(cxxopts::ParseResult &options) {
 
             memcpy(file + part * options["mtu"].as<int>(), buffer + 16, size);
         } else if (strncmp(buffer, "FINISH", 6) == 0) {
-            std::cout << "Server finish transfering" << std::endl;
-
-            std::cout << "Create thread, witch ask not accepted parts" << std::endl;
-            std::thread(checkParts).detach();
+            // If we don't receive a finish message
+            if (!finish) {
+                std::cout << "Server finish transfering" << std::endl;
+                finish = true;
+            }
         }
     }
     delete[] buffer;

src/Sender.cpp

@@ -26,9 +26,6 @@ void sendPart(int part_index) {
 }
 
 void run(cxxopts::ParseResult &options) {
-    std::mt19937 gen(time(0));                    // TODO: remove (for tests only)
-    std::uniform_int_distribution<> uid(0, 100);  // TODO: remove (for tests only)
-
     buffer = new char[2 * mtu];
 
     std::ifstream input(fileName, std::ios::binary);        //
@@ -45,46 +42,41 @@ void run(cxxopts::ParseResult &options) {
 
 
     snprintf(buffer, 11, "NEW_PACKET");                            //
-    Utils::writeBytesFromInt(buffer + 10, file_length, 4);         // Send information
+    Utils::writeBytesFromInt(buffer + 10, file_length, 4);         //
+    Utils::writeBytesFromInt(buffer + 14, mtu, 4);                 // Send information
     sendto(_socket, buffer, 14, 0, (sockaddr*) &broadcast_address, // about new file
-           sizeof(broadcast_address));                             //
-    std::cout << "Send information about new file with size " << (int) file_length << std::endl;
+           sizeof(broadcast_address));                             // + size and mtu
+    std::cout << "Send information about new file with size ";     //
+    std::cout << (int) file_length << " mtu " << mtu <<std::endl;  //
 
     int part_index = 0;
 
     while (part_index * mtu < file_length) {  //
-        sent_part.insert({ part_index, 0 });  //
-                                              //
-        if (uid(gen) < 90) {                  // Send all parts every 1 sec
-            sendPart(part_index);             //
-        }                                     //
-                                              //
+        sent_part.insert({ part_index, 0 });  // 
+                                              // 
+        sendPart(part_index);                 // Send parts 
+                                              //   every 20ms
         part_index++;                         //
-        std::this_thread::sleep_for(1s);      //
+        std::this_thread::sleep_for(20ms);    //
     }                                         //
 
     snprintf(buffer, 7, "FINISH");                                // Send information
     sendto(_socket, buffer, 6, 0, (sockaddr*) &broadcast_address, // the end of transaction
            sizeof(broadcast_address));                            //
     std::cout << "File transfer complete" << std::endl;
+    // Last time, while sender send message about finish
+    long lastFinishSendTime = 0;
 
-    // Set socket receive timeout to 1 sec
-    #if defined(_WIN32) || defined(_WIN64)
-    int tv = 1 * 1000;  // user timeout in milliseconds [ms]
-    setsockopt(_socket, SOL_SOCKET, SO_RCVTIMEO, (char*)&tv, sizeof(tv));
-    #else
-    struct timeval tv;
-    tv.tv_sec = 1;
-    tv.tv_usec = 0;
-    setsockopt(_socket, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof(tv));
-    #endif
-
-    while (ttl--) {
+    while (ttl) {
         auto result = recvfrom(_socket, (char *)buffer, 100, 0, (struct sockaddr*) &broadcast_address, &client_address_length);
-        std::cout << result << std::endl;
-        if (result <= 0) continue;
 
-        ttl = ttl_max;
+        // Every seconds send information about end of sending file
+        if (result <= 0) {
+            ttl--;
+            snprintf(buffer, 7, "FINISH");
+            sendto(_socket, buffer, 6, 0, (struct sockaddr*) &broadcast_address, sizeof(broadcast_address));
+            continue;
+        }
 
         if (strncmp(buffer, "RESEND", 6) == 0) {
             int part = Utils::getIntFromBytes(buffer + 6, 4);
@@ -93,22 +85,27 @@ void run(cxxopts::ParseResult &options) {
             auto now_ms   = std::chrono::time_point_cast<std::chrono::seconds>(now);
             auto epoch    = now_ms.time_since_epoch();
             auto value    = std::chrono::duration_cast<std::chrono::seconds>(epoch);
-            long duration = value.count();
+            long duration = value.count(); // Unix time in second
+
+            ttl = ttl_max;
 
-            if (duration - sent_part[part] > 1) {
+            if (duration - sent_part[part] >= 1) {
                 sent_part[part] = duration;
                 std::cout << "Клиент запросил " << part << " часть" << std::endl;
                 sendPart(part);
             }
-        } else if (strncmp(buffer, "STATUS", 6) == 0) {
-            std::cout << "Клиент запросил статус" << std::endl;
 
-            snprintf(buffer, 6, "FINISH");
-            sendto(_socket, buffer, 6, 0, (struct sockaddr*) &broadcast_address, sizeof(broadcast_address));
+            // Every seconds send information about end of sending file
+            if (duration - lastFinishSendTime >= 1) {
+                lastFinishSendTime = duration;
+                snprintf(buffer, 7, "FINISH");
+                sendto(_socket, buffer, 6, 0, (struct sockaddr*) &broadcast_address, sizeof(broadcast_address));
+            }
         }
+
     }
     std::cout << "Process no longer be working" << std::endl;
 }
 
 
-} //namespace Sender
+} //namespace Sender

src/Utils.hpp

@@ -2,6 +2,7 @@
 #define FILEBROADCASTER_UTILS_H
 
 #if defined(_WIN32) || defined(_WIN64)
+#define _WINSOCK_DEPRECATED_NO_WARNINGS
 #define addr_len int                    //
 #include <winsock2.h>                   // Windows
 #include <windows.h>                    //  socket
@@ -11,7 +12,6 @@
 #define SOCKADDR_IN sockaddr_in         //
 #define addr_len socklen_t              // Linux socket
 #include <arpa/inet.h>                  //
-#include <pthread.h>                    //
 #endif
 
 #include <thread>
@@ -33,9 +33,6 @@
 #include "../lib/cxxopts/include/cxxopts.hpp"
 #include "Config.hpp"
 
-#define _WINSOCK_DEPRECATED_NO_WARNINGS
-#define async
-
 
 using namespace std::chrono_literals;
 
@@ -67,18 +64,6 @@ namespace Utils {
             buffer[count - i - 1] = (char) (value >> (i * 8));
         }
     }
-
-    /**
-     * Run in secondary thread
-     * Decrease ttl every second
-     * If ttl < 0 program should stop
-     */
-    void async checkTTL() {
-        while (ttl--) {
-            std::this_thread::sleep_for(1s);
-        }
-        std::cout << "Process no longer be working" << std::endl;
-    }
 }
 
 #endif //FILEBROADCASTER_UTILS_H