wdt cygwin found problems

Summary: fixes facebook#114

Reviewed By: uddipta

Differential Revision: D2999244

fb-gh-sync-id: 1a9fca5778aaf42c359326fad4a3857de025e2f4
shipit-source-id: 1a9fca5778aaf42c359326fad4a3857de025e2f4

.clang-format

@@ -6,4 +6,5 @@ AllowShortFunctionsOnASingleLine: false
 #IndentWidth: 4
 UseTab: Never
 Standard: Cpp11
+SortIncludes: false
 ...

README.md

@@ -308,15 +308,20 @@ wdt_max_send_test.sh
 (facebook only:)
 Make sure to do the following, before "arc diff":
 ```
- (cd wdt ; ./build/clangformat.sh ; ./build/version_update.tcl )
+ (cd wdt ; ./build/clangformat.sh )
+ # if you changed the minor version of the protocol (in CMakeLists.txt)
+ # run (cd wdt ; ./build/version_update.tcl ) to sync with fbcode's WdtConfig.h
 
- fbconfig  --clang --with-project-version clang:dev -r  wdt
+ fbconfig  --clang --sanitize=address -r  wdt
 
  fbmake runtests --run-disabled --extended-tests
+ # Optionally: opt build
  fbmake runtests_opt
  fbmake opt
-
+ # Sender max speed test
  wdt/test/wdt_max_send_test.sh
+ # Check buck build
+ buck build wdt/...
 ```
 
 and check the output of the last step to make sure one of the 3 runs is

ReceiverThread.cpp

@@ -81,12 +81,18 @@ int64_t readAtMost(ServerSocket &s, char *buf, int64_t max, int64_t atMost) {
 }
 
 const ReceiverThread::StateFunction ReceiverThread::stateMap_[] = {
-    &ReceiverThread::listen, &ReceiverThread::acceptFirstConnection,
-    &ReceiverThread::acceptWithTimeout, &ReceiverThread::sendLocalCheckpoint,
-    &ReceiverThread::readNextCmd, &ReceiverThread::processFileCmd,
-    &ReceiverThread::processSettingsCmd, &ReceiverThread::processDoneCmd,
-    &ReceiverThread::processSizeCmd, &ReceiverThread::sendFileChunks,
-    &ReceiverThread::sendGlobalCheckpoint, &ReceiverThread::sendDoneCmd,
+    &ReceiverThread::listen,
+    &ReceiverThread::acceptFirstConnection,
+    &ReceiverThread::acceptWithTimeout,
+    &ReceiverThread::sendLocalCheckpoint,
+    &ReceiverThread::readNextCmd,
+    &ReceiverThread::processFileCmd,
+    &ReceiverThread::processSettingsCmd,
+    &ReceiverThread::processDoneCmd,
+    &ReceiverThread::processSizeCmd,
+    &ReceiverThread::sendFileChunks,
+    &ReceiverThread::sendGlobalCheckpoint,
+    &ReceiverThread::sendDoneCmd,
     &ReceiverThread::sendAbortCmd,
     &ReceiverThread::waitForFinishOrNewCheckpoint,
     &ReceiverThread::finishWithError};

Reporting.cpp

@@ -313,9 +313,21 @@ void ProgressReporter::logProgress(int64_t effectiveDataBytes, int progress,
 }
 
 const std::string PerfStatReport::statTypeDescription_[] = {
-    "Socket Read", "Socket Write", "File Open", "File Close", "File Read",
-    "File Write", "Sync File Range", "fsync", "File Seek", "Throttler Sleep",
-    "Receiver Wait Sleep", "Directory creation", "Ioctl", "Unlink", "Fadvise"};
+    "Socket Read",
+    "Socket Write",
+    "File Open",
+    "File Close",
+    "File Read",
+    "File Write",
+    "Sync File Range",
+    "fsync",
+    "File Seek",
+    "Throttler Sleep",
+    "Receiver Wait Sleep",
+    "Directory creation",
+    "Ioctl",
+    "Unlink",
+    "Fadvise"};
 
 PerfStatReport::PerfStatReport(const WdtOptions& options) {
   static_assert(

Reporting.h

@@ -511,7 +511,7 @@ class PerfStatReport {
     FILE_READ,
     FILE_WRITE,
     SYNC_FILE_RANGE,
-    FSYNC,
+    FSYNC_STATS,  // just 'FSYNC' is defined on Windows/conflicts
     FILE_SEEK,
     THROTTLER_SLEEP,
     RECEIVER_WAIT_SLEEP,  // receiver sleep duration between sending wait cmd to

Sender.cpp

@@ -373,7 +373,8 @@ ErrorCode Sender::start() {
       dirQueue_->enableFileDeletion();
     } else {
       LOG(WARNING) << "Turning off extra file deletion on the receiver side "
-                      "because of protocol version " << protocolVersion_;
+                      "because of protocol version "
+                   << protocolVersion_;
     }
   }
   dirThread_ = dirQueue_->buildQueueAsynchronously();

SenderThread.cpp

@@ -26,12 +26,12 @@ std::ostream &operator<<(std::ostream &os, const SenderThread &senderThread) {
 }
 
 const SenderThread::StateFunction SenderThread::stateMap_[] = {
-    &SenderThread::connect, &SenderThread::readLocalCheckPoint,
-    &SenderThread::sendSettings, &SenderThread::sendBlocks,
-    &SenderThread::sendDoneCmd, &SenderThread::sendSizeCmd,
-    &SenderThread::checkForAbort, &SenderThread::readFileChunks,
+    &SenderThread::connect,         &SenderThread::readLocalCheckPoint,
+    &SenderThread::sendSettings,    &SenderThread::sendBlocks,
+    &SenderThread::sendDoneCmd,     &SenderThread::sendSizeCmd,
+    &SenderThread::checkForAbort,   &SenderThread::readFileChunks,
     &SenderThread::readReceiverCmd, &SenderThread::processDoneCmd,
-    &SenderThread::processWaitCmd, &SenderThread::processErrCmd,
+    &SenderThread::processWaitCmd,  &SenderThread::processErrCmd,
     &SenderThread::processAbortCmd, &SenderThread::processVersionMismatch};
 
 std::unique_ptr<ClientSocket> SenderThread::connectToReceiver(
@@ -500,8 +500,8 @@ SenderState SenderThread::readFileChunks() {
       // fileChunksInfoList. Number of chunks we decode should match with the
       // number mentioned in the Chunks cmd.
       LOG(ERROR) << "Number of file chunks received is more than the number "
-                    "mentioned in CHUNKS_CMD " << numFileChunks << " "
-                 << numFiles;
+                    "mentioned in CHUNKS_CMD "
+                 << numFileChunks << " " << numFiles;
       threadStats_.setLocalErrorCode(PROTOCOL_ERROR);
       return END;
     }
@@ -588,8 +588,8 @@ ErrorCode SenderThread::readNextReceiverCmd() {
       return SOCKET_READ_ERROR;
     }
     LOG(INFO) << "Read receiver command failed, but number of unacked "
-                 "bytes decreased, retrying socket read " << numUnackedBytes
-              << " " << curUnackedBytes;
+                 "bytes decreased, retrying socket read "
+              << numUnackedBytes << " " << curUnackedBytes;
     numUnackedBytes = curUnackedBytes;
   }
   // we are assuming that sender and receiver tcp buffer sizes are same. So, we

build/BUILD.md

@@ -9,6 +9,9 @@ See also [README.md](../README.md#dependencies) Dependencies section
 Checkout the .travis.yml and travis_linux.sh and travis_osx.sh
 for build bootstrap without root/sudo requirement
 
+We don't have yet a native Windows port (please contribute!) but it does
+build and runs with Cygwin64 using the Linux instructions (and static linking)
+
 # Notes:
  On Ubuntu 14.04 - to get g++ 4.9
  ```

util/DirectorySourceQueue.cpp

@@ -99,8 +99,8 @@ void DirectorySourceQueue::setFollowSymlinks(const bool followSymlinks) {
   }
 }
 
-std::vector<SourceMetaData *> &
-DirectorySourceQueue::getDiscoveredFilesMetaData() {
+std::vector<SourceMetaData *>
+    &DirectorySourceQueue::getDiscoveredFilesMetaData() {
   return sharedFileData_;
 }
 

util/DirectorySourceQueue.h

@@ -348,7 +348,8 @@ class DirectorySourceQueue : public SourceQueue {
    */
   std::priority_queue<std::unique_ptr<ByteSource>,
                       std::vector<std::unique_ptr<ByteSource>>,
-                      SourceComparator> sourceQueue_;
+                      SourceComparator>
+      sourceQueue_;
 
   /// Transfer stats for sources which are not transferred
   std::vector<TransferStats> failedSourceStats_;

util/FileCreator.cpp

@@ -144,8 +144,10 @@ int FileCreator::openForBlocks(ThreadCtx &threadCtx,
   auto it = fileStatusMap_.find(blockDetails->seqId);
   if (blockDetails->allocationStatus == EXISTS_CORRECT_SIZE &&
       it == fileStatusMap_.end()) {
-    it = fileStatusMap_.insert(std::make_pair(blockDetails->seqId,
-                                              FileCreator::ALLOCATED)).first;
+    it =
+        fileStatusMap_
+            .insert(std::make_pair(blockDetails->seqId, FileCreator::ALLOCATED))
+            .first;
   }
   if (it == fileStatusMap_.end()) {
     // allocation has not started for this file

util/FileWriter.cpp

@@ -85,7 +85,7 @@ ErrorCode FileWriter::write(char *buf, int64_t size) {
             << " for file " << blockDetails_->fileName;
     bool finished = ((totalWritten_ + size) == blockDetails_->dataSize);
     if (finished && options.isLogBasedResumption()) {
-      PerfStatCollector statCollector(threadCtx_, PerfStatReport::FSYNC);
+      PerfStatCollector statCollector(threadCtx_, PerfStatReport::FSYNC_STATS);
       if (fsync(fd_) != 0) {
         PLOG(ERROR) << "fsync failed for " << blockDetails_->fileName
                     << " offset " << blockDetails_->offset << " file-size "

util/ThreadTransferHistory.cpp

@@ -140,8 +140,8 @@ ErrorCode ThreadTransferHistory::validateCheckpoint(
   }
   if (checkpoint.numBlocks < lastCheckpoint_->numBlocks) {
     LOG(ERROR) << "Current checkpoint must be higher than previous checkpoint, "
-                  "Last checkpoint: " << *lastCheckpoint_
-               << ", Current checkpoint: " << checkpoint;
+                  "Last checkpoint: "
+               << *lastCheckpoint_ << ", Current checkpoint: " << checkpoint;
     return INVALID_CHECKPOINT;
   }
   if (checkpoint.numBlocks > lastCheckpoint_->numBlocks) {

util/TransferLogManager.cpp

@@ -29,7 +29,8 @@ const int TransferLogManager::LOG_VERSION = 2;
 int64_t LogEncoderDecoder::timestampInMicroseconds() const {
   auto timestamp = Clock::now();
   return std::chrono::duration_cast<std::chrono::microseconds>(
-             timestamp.time_since_epoch()).count();
+             timestamp.time_since_epoch())
+      .count();
 }
 
 int64_t LogEncoderDecoder::encodeLogHeader(char *dest,
@@ -239,12 +240,31 @@ ErrorCode TransferLogManager::openLog() {
   WDT_CHECK(!rootDir_.empty()) << "Root directory not set";
   WDT_CHECK(options_.enable_download_resumption);
 
-  int openFlags = O_CREAT | O_RDWR;
   const std::string logPath = getFullPath(kWdtLogName);
-  fd_ = ::open(logPath.c_str(), openFlags, 0644);
+  fd_ = ::open(logPath.c_str(), O_RDWR);
   if (fd_ < 0) {
-    PLOG(ERROR) << "Could not open wdt log " << logPath;
-    return BYTE_SOURCE_READ_ERROR;
+    if (errno != ENOENT) {
+      PLOG(ERROR) << "Could not open wdt log " << logPath;
+      return TRANSFER_LOG_ACQUIRE_ERROR;
+    } else {
+      // creation of the log path (which can still be a race)
+      LOG(INFO) << logPath << " doesn't exist... creating...";
+      fd_ = ::open(logPath.c_str(), O_CREAT | O_EXCL, 0644);
+      if (fd_ < 0) {
+        PLOG(WARNING) << "Could not create wdt log (maybe ok if race): "
+                      << logPath;
+      } else {
+        // On windows/cygwin for instance the flock will silently succeed yet
+        // not lock on a newly created file... workaround is to close and reopen
+        ::close(fd_);
+      }
+      fd_ = ::open(logPath.c_str(), O_RDWR);
+      if (fd_ < 0) {
+        PLOG(ERROR) << "Still couldn't open wdt log after create attempt: "
+                    << logPath;
+        return TRANSFER_LOG_ACQUIRE_ERROR;
+      }
+    }
   }
   // try to acquire file lock
   if (::flock(fd_, LOCK_EX | LOCK_NB) != 0) {
@@ -380,7 +400,8 @@ bool TransferLogManager::verifySenderIp(const std::string &curSenderIp) {
                 << curSenderIp;
     } else if (senderIp_ != curSenderIp) {
       LOG(ERROR) << "Current sender ip does not match ip in the "
-                    "transfer log " << curSenderIp << " " << senderIp_
+                    "transfer log "
+                 << curSenderIp << " " << senderIp_
                  << ", ignoring transfer log";
       verifySuccessful = false;
       invalidateDirectory();
@@ -577,7 +598,8 @@ bool LogParser::writeFileInvalidationEntries(int fd,
     int written = ::write(fd, buf, size);
     if (written != size) {
       PLOG(ERROR) << "Disk write error while writing invalidation entry to "
-                     "transfer log " << written << " " << size;
+                     "transfer log "
+                  << written << " " << size;
       return false;
     }
   }
@@ -691,8 +713,8 @@ ErrorCode LogParser::processFileCreationEntry(char *buf, int size) {
   }
   if (options_.resume_using_dir_tree) {
     LOG(ERROR) << "Can not have a file creation entry in directory based "
-                  "resumption mode " << fileName << " " << seqId << " "
-               << fileSize;
+                  "resumption mode "
+               << fileName << " " << seqId << " " << fileSize;
     return INVALID_LOG;
   }
   if (fileInfoMap_.find(seqId) != fileInfoMap_.end() ||
@@ -747,7 +769,8 @@ ErrorCode LogParser::processFileResizeEntry(char *buf, int size) {
   }
   if (options_.resume_using_dir_tree) {
     LOG(ERROR) << "Can not have a file resize entry in directory based "
-                  "resumption mode " << seqId << " " << fileSize;
+                  "resumption mode "
+               << seqId << " " << fileSize;
     return INVALID_LOG;
   }
   auto it = fileInfoMap_.find(seqId);
@@ -796,8 +819,8 @@ ErrorCode LogParser::processBlockWriteEntry(char *buf, int size) {
   }
   if (options_.resume_using_dir_tree) {
     LOG(ERROR) << "Can not have a block write entry in directory based "
-                  "resumption mode " << seqId << " " << offset << " "
-               << blockSize;
+                  "resumption mode "
+               << seqId << " " << offset << " " << blockSize;
     return INVALID_LOG;
   }
   if (invalidSeqIds_.find(seqId) != invalidSeqIds_.end()) {
@@ -842,7 +865,8 @@ ErrorCode LogParser::processFileInvalidationEntry(char *buf, int size) {
   }
   if (options_.resume_using_dir_tree) {
     LOG(ERROR) << "Can not have a file invalidation entry in directory based "
-                  "resumption mode " << seqId;
+                  "resumption mode "
+               << seqId;
     return INVALID_LOG;
   }
   if (fileInfoMap_.find(seqId) == fileInfoMap_.end() &&

util/WdtSocket.cpp

@@ -150,7 +150,8 @@ int WdtSocket::readWithTimeout(char *buf, int nbyte, int timeoutMs,
   if (supportUnencryptedPeer_ && readErrorCode_ == UNEXPECTED_CMD_ERROR) {
     LOG(WARNING)
         << "Turning off encryption since the other side does not support "
-           "encryption " << port_;
+           "encryption "
+        << port_;
     readErrorCode_ = OK;
     buf[0] = buf_[0];
     numRead = 1;
@@ -517,7 +518,7 @@ int WdtSocket::getUnackedBytes() const {
 }
 WdtSocket::~WdtSocket() {
   VLOG(1) << "~WdtSocket " << port_ << " " << fd_;
-  closeConnection();
+  closeNoCheck();
 }
 }
 }