win: refactor some code in RequestContext

Summary:
Previously, the notification callback code was pretty ad-hoc in how it dealt
with the request context and handling asynchronous callbacks, in order to share
more code with FUSE, let's add a catchErrors method to the PrjfsRequestContext
similarly to what is done in the FUSE code. Once timeouts and notifications
will be added, the catchErrors code will be moved into the parent class and all
of this code will be common between ProjectedFS and FUSE.

Reviewed By: fanzeyi

Differential Revision: D23626748

fbshipit-source-id: 70fae3d4a276be374f58559cc1fb05c8e56e5c2d

eden/fs/win/mount/PrjfsChannel.cpp

@@ -251,43 +251,34 @@ HRESULT notification(
       auto dispatcher = channel->getDispatcher();
       auto context =
           std::make_unique<PrjfsRequestContext>(channel, *callbackData);
-      auto requestWatch =
-          std::shared_ptr<RequestMetricsScope::LockedRequestWatchList>(nullptr);
+      auto histogram = it->second.histogram;
       auto handler = it->second.handler;
 
       auto relPath = RelativePath(callbackData->FilePathName);
       auto destPath = RelativePath(destinationFileName);
 
-      context->startRequest(
-          dispatcher->getStats(), it->second.histogram, requestWatch);
-      auto fut =
-          (dispatcher->*handler)(relPath, destPath, isDirectory, *context);
+      context
+          ->catchErrors(folly::makeFutureWith([context = context.get(),
+                                               handler = handler,
+                                               histogram = histogram,
+                                               dispatcher = dispatcher,
+                                               relPath = std::move(relPath),
+                                               destPath = std::move(destPath),
+                                               isDirectory] {
+            auto requestWatch =
+                std::shared_ptr<RequestMetricsScope::LockedRequestWatchList>(
+                    nullptr);
+            context->startRequest(
+                dispatcher->getStats(), histogram, requestWatch);
 
-      std::move(fut).thenTryInline(
-          [context = std::move(context),
-           channel = channel](folly::Try<folly::Unit>&& try_) {
-            SCOPE_EXIT {
-              context->finishRequest();
-            };
-
-            HRESULT result = S_OK;
-            if (try_.hasException()) {
-              auto* err = try_.tryGetExceptionObject<std::exception>();
-              DCHECK(err);
-              result = exceptionToHResult(*err);
-            }
-
-            PRJ_COMPLETE_COMMAND_EXTENDED_PARAMETERS extra{};
-            extra.CommandType = PRJ_COMPLETE_COMMAND_TYPE_NOTIFICATION;
-            result = PrjCompleteCommand(
-                channel->getMountChannelContext(),
-                context->getCommandId(),
-                result,
-                &extra);
-            if (FAILED(result)) {
-              XLOG(ERR) << "Couldn't complete command";
-            }
-          });
+            return (dispatcher->*handler)(
+                       relPath, destPath, isDirectory, *context)
+                .thenValue([context = context](auto&&) {
+                  context->sendNotificationSuccess();
+                });
+          }))
+          // Make sure that the context is alive for the duration of the future.
+          .ensure([context = std::move(context)] {});
 
       return HRESULT_FROM_WIN32(ERROR_IO_PENDING);
     }
@@ -450,5 +441,29 @@ void PrjfsChannel::flushNegativePathCache() {
   }
 }
 
+namespace {
+void sendReply(
+    PRJ_NAMESPACE_VIRTUALIZATION_CONTEXT context,
+    int32_t commandId,
+    HRESULT result,
+    PRJ_COMPLETE_COMMAND_EXTENDED_PARAMETERS* FOLLY_NULLABLE extra) {
+  result = PrjCompleteCommand(context, commandId, result, extra);
+  if (FAILED(result)) {
+    XLOG(ERR) << "Couldn't complete command: " << commandId << ": "
+              << win32ErrorToString(result);
+  }
+}
+} // namespace
+
+void PrjfsChannel::sendSuccess(
+    int32_t commandId,
+    PRJ_COMPLETE_COMMAND_EXTENDED_PARAMETERS* FOLLY_NULLABLE extra) {
+  sendReply(getMountChannelContext(), commandId, S_OK, extra);
+}
+
+void PrjfsChannel::sendError(int32_t commandId, HRESULT result) {
+  sendReply(getMountChannelContext(), commandId, result, nullptr);
+}
+
 } // namespace eden
 } // namespace facebook

eden/fs/win/mount/PrjfsChannel.h

@@ -60,6 +60,12 @@ class PrjfsChannel : public FsChannel {
     return mountChannel_;
   }
 
+  void sendSuccess(
+      int32_t commandId,
+      PRJ_COMPLETE_COMMAND_EXTENDED_PARAMETERS* FOLLY_NULLABLE extra);
+
+  void sendError(int32_t commandId, HRESULT error);
+
  private:
   //
   // Channel to talk to projectedFS.

eden/fs/win/mount/PrjfsRequestContext.h

@@ -28,11 +28,35 @@ class PrjfsRequestContext : public RequestContext {
         channel_(channel),
         commandId_(prjfsData.CommandId) {}
 
-  int32_t getCommandId() const {
-    return commandId_;
+  folly::Future<folly::Unit> catchErrors(folly::Future<folly::Unit>&& fut) {
+    return std::move(fut).thenTryInline([this](folly::Try<folly::Unit>&& try_) {
+      SCOPE_EXIT {
+        finishRequest();
+      };
+
+      if (try_.hasException()) {
+        auto* err = try_.tryGetExceptionObject<std::exception>();
+        DCHECK(err);
+        sendError(exceptionToHResult(*err));
+      }
+    });
+  }
+
+  void sendSuccess() const {
+    return channel_->sendSuccess(commandId_, nullptr);
+  }
+
+  void sendNotificationSuccess() const {
+    PRJ_COMPLETE_COMMAND_EXTENDED_PARAMETERS extra{};
+    extra.CommandType = PRJ_COMPLETE_COMMAND_TYPE_NOTIFICATION;
+    return channel_->sendSuccess(commandId_, &extra);
   }
 
  private:
+  void sendError(HRESULT result) const {
+    return channel_->sendError(commandId_, result);
+  }
+
   PrjfsChannel* channel_;
   int32_t commandId_;
 };