Wdt State machine refactor

Summary: Get rid of ThreadData and move different functionalities into the appropriate modules

Reviewed By: @uddipta

Differential Revision: D2445473

Author: nikunjy

CMakeLists.txt

@@ -22,7 +22,7 @@ cmake_minimum_required(VERSION 3.2)
 # There is no C per se in WDT but if you use CXX only here many checks fail
 # Version is Major.Minor.YYMMDDX for up to 10 releases per day
 # Minor currently is also the protocol version - has to match with Protocol.cpp
-project("WDT" LANGUAGES C CXX VERSION 1.21.1510120)
+project("WDT" LANGUAGES C CXX VERSION 1.22.1510210)
 
 # On MacOS this requires the latest (master) CMake (and/or CMake 3.1.1/3.2)
 set(CMAKE_CXX_STANDARD 11)
@@ -79,8 +79,13 @@ ErrorCodes.cpp
 FileByteSource.cpp
 FileCreator.cpp
 Protocol.cpp
+WdtThread.cpp
+ThreadsController.cpp
+ReceiverThread.cpp
 Receiver.cpp
 Reporting.cpp
+ThreadTransferHistory.cpp
+SenderThread.cpp
 Sender.cpp
 ServerSocket.cpp
 SocketUtils.cpp

ErrorCodes.h

@@ -9,7 +9,7 @@
 #pragma once
 
 #include <wdt/WdtConfig.h>
-
+#include <glog/logging.h>
 #include <string>
 
 namespace facebook {

FileCreator.cpp

@@ -164,8 +164,7 @@ int FileCreator::openExistingFile(const string &relPathStr) {
   WDT_CHECK(relPathStr[0] != '/');
   WDT_CHECK(relPathStr.back() != '/');
 
-  string path(rootDir_);
-  path.append(relPathStr);
+  const string path = rootDir_ + relPathStr;
 
   int openFlags = O_WRONLY;
   START_PERF_TIMER
@@ -184,8 +183,7 @@ int FileCreator::createFile(const string &relPathStr) {
   CHECK(relPathStr[0] != '/');
   CHECK(relPathStr.back() != '/');
 
-  std::string path(rootDir_);
-  path.append(relPathStr);
+  const string path = rootDir_ + relPathStr;
 
   int p = relPathStr.size();
   while (p && relPathStr[p - 1] != '/') {

Protocol.cpp

@@ -49,7 +49,8 @@ int Protocol::negotiateProtocol(int requestedProtocolVersion,
 }
 
 std::ostream &operator<<(std::ostream &os, const Checkpoint &checkpoint) {
-  os << "num-blocks: " << checkpoint.numBlocks
+  os << "checkpoint-port: " << checkpoint.port
+     << "num-blocks: " << checkpoint.numBlocks
      << " seq-id: " << checkpoint.lastBlockSeqId
      << " block-offset: " << checkpoint.lastBlockOffset
      << " received-bytes: " << checkpoint.lastBlockReceivedBytes;

README.md

@@ -125,6 +125,10 @@ caller get the mutable object of options and set different options accordingly.
 When wdt is run in a standalone mode, behavior is changed through gflags in
 wdtCmdLine.cpp
 
+* WdtThread.{h|cpp}
+Common functionality and settings between SenderThread and ReceiverThread.
+Both of these kind of threads inherit from this base class.
+
 * WdtBase.{h|cpp}
 
 Common functionality and settings between Sender and Receiver
@@ -156,22 +160,36 @@ directory, sorted by decreasing size (as they are discovered, you can start
 pulling from the queue even before all the files are found, it will return
 the current largest file)
 
+* ThreadTransferHistory.{h|cpp}
 
-* Sender.{h|cpp}
+Every thread maintains a transfer history so that when a connection breaks
+it can talk to the receiver to find out up to where in the history has been 
+sent. This class encapsulates all the logic for that bookkeeping
+
+* SenderThread.{h|cpp}
+
+Implements the functionality of one sender thread, which binds to a certain port
+and sends files over.
 
-Formerly wdtlib.cpp - main code sending files
+* Sender.{h|cpp}
 
+Spawns multiple SenderThread threads and sends the data across to receiver
 
 ### Consuming / Receiving
 
 * FileCreator.{h|cpp}
 
 Creates file and directories necessary for said file (mkdir -p like)
 
-* Receiver.{h|cpp}
+* ReceiverThread.{h|cpp}
 
-Formerly wdtlib.cpp - main code receiving files
+Implements the funcionality of the receiver threads, responsible for listening on
+a port and receiving files over the network.
+
+* Receiver.{h|cpp}
 
+Parent receiver class that spawns multiple ReceiverThread threads and receives
+data from a remote host
 
 ### Low level building blocks