[SE] Make Stream movable

[jhen]

Summary

The example code makes it clear that this is a much better design
decision.

[jlebar]

Yeah, I like this better too.

streamexecutor/include/streamexecutor/Stream.h

344 ↗

(On Diff #70022)

No longer mutable

streamexecutor/lib/Stream.cpp

21 ↗

(On Diff #70022)

Nit, usually prefer make_unique so that it's clear that the destination variable is a unique_ptr.

[jhen]

• Respond to jlebar's comments

streamexecutor/include/streamexecutor/Stream.h

344 ↗

(On Diff #70022)

Removed the comment

parallel-libs/trunk/streamexecutor/examples/Example.cpp

int main() {
 
   // Allocate memory on the device.
   se::GlobalDeviceMemory<float> X =
       getOrDie(Device->allocateDeviceMemory<float>(ArraySize));
   se::GlobalDeviceMemory<float> Y =
       getOrDie(Device->allocateDeviceMemory<float>(ArraySize));
 
   // Run operations on a stream.
-  std::unique_ptr<se::Stream> Stream = getOrDie(Device->createStream());
-  Stream->thenCopyH2D<float>(HostX, X)
+  se::Stream Stream = getOrDie(Device->createStream());
+  Stream.thenCopyH2D<float>(HostX, X)
       .thenCopyH2D<float>(HostY, Y)
       .thenLaunch(ArraySize, 1, *Kernel, A, X, Y)
       .thenCopyD2H<float>(X, HostX);
   // Wait for the stream to complete.
-  se::dieIfError(Stream->blockHostUntilDone());
+  se::dieIfError(Stream.blockHostUntilDone());
 
   // Process output data in HostX.
   std::vector<float> ExpectedX = {4, 47, 90, 133};
   for (size_t I = 0; I < ArraySize; ++I) {
     assert(HostX[I] == ExpectedX[I]);
   }
 
   // Free device memory.
   se::dieIfError(Device->freeDeviceMemory(X));
   se::dieIfError(Device->freeDeviceMemory(Y));
   /// [Example saxpy host main]
 }

parallel-libs/trunk/streamexecutor/include/streamexecutor/Device.h

Expected<std::unique_ptr<PlatformKernelHandle>> MaybeKernelHandle =
         PDevice->createKernel(Spec);
     if (!MaybeKernelHandle) {
       return MaybeKernelHandle.takeError();
     }
     return llvm::make_unique<KernelT>(Spec.getKernelName(),
                                       std::move(*MaybeKernelHandle));
   }
 
-  Expected<std::unique_ptr<Stream>> createStream();
+  /// Creates a stream object for this device.
+  Expected<Stream> createStream();
 
   /// Allocates an array of ElementCount entries of type T in device memory.
   template <typename T>
   Expected<GlobalDeviceMemory<T>> allocateDeviceMemory(size_t ElementCount) {
     Expected<GlobalDeviceMemoryBase> MaybeBase =
         PDevice->allocateDeviceMemory(ElementCount * sizeof(T));
     if (!MaybeBase)
       return MaybeBase.takeError();

parallel-libs/trunk/streamexecutor/include/streamexecutor/Stream.h

 /// If any given operation fails when entraining work for the stream, isOK()
 /// will indicate that an error has occurred and getStatus() will get the first
 /// error that occurred on the stream. There is no way to clear the error state
 /// of a stream once it is in an error state.
 class Stream {
 public:
   explicit Stream(std::unique_ptr<PlatformStreamHandle> PStream);
 
+  Stream(Stream &&Other) = default;
+  Stream &operator=(Stream &&Other) = default;
+
   ~Stream();
 
   /// Returns whether any error has occurred while entraining work on this
   /// stream.
   bool isOK() const {
-    llvm::sys::ScopedReader ReaderLock(ErrorMessageMutex);
+    llvm::sys::ScopedReader ReaderLock(*ErrorMessageMutex);
     return !ErrorMessage;
   }
 
   /// Returns the status created by the first error that occurred while
   /// entraining work on this stream.
   Error getStatus() const {
-    llvm::sys::ScopedReader ReaderLock(ErrorMessageMutex);
+    llvm::sys::ScopedReader ReaderLock(*ErrorMessageMutex);
     if (ErrorMessage)
       return make_error(*ErrorMessage);
     else
       return Error::success();
   }
 
   // Blocks the calling host thread until all work enqueued on this Stream
   // completes.
public:
   ///@} End device memory copying functions
 
 private:
   /// Sets the error state from an Error object.
   ///
   /// Does not overwrite the error if it is already set.
   void setError(Error &&E) {
     if (E) {
-      llvm::sys::ScopedWriter WriterLock(ErrorMessageMutex);
+      llvm::sys::ScopedWriter WriterLock(*ErrorMessageMutex);
       if (!ErrorMessage)
         ErrorMessage = consumeAndGetMessage(std::move(E));
     }
   }
 
   /// Sets the error state from an error message.
   ///
   /// Does not overwrite the error if it is already set.
   void setError(llvm::Twine Message) {
-    llvm::sys::ScopedWriter WriterLock(ErrorMessageMutex);
+    llvm::sys::ScopedWriter WriterLock(*ErrorMessageMutex);
     if (!ErrorMessage)
       ErrorMessage = Message.str();
   }
 
   /// The PlatformDevice that supports the operations of this stream.
   PlatformDevice *PDevice;
 
   /// The platform-specific stream handle for this instance.
   std::unique_ptr<PlatformStreamHandle> ThePlatformStream;
 
   /// Mutex that guards the error state flags.
-  ///
-  /// Mutable so that it can be obtained via const reader lock.
-  mutable llvm::sys::RWMutex ErrorMessageMutex;
+  std::unique_ptr<llvm::sys::RWMutex> ErrorMessageMutex;
 
   /// First error message for an operation in this stream or empty if there have
   /// been no errors.
   llvm::Optional<std::string> ErrorMessage;
 
   Stream(const Stream &) = delete;
   void operator=(const Stream &) = delete;
 };
 
 } // namespace streamexecutor
 
 #endif // STREAMEXECUTOR_STREAM_H

parallel-libs/trunk/streamexecutor/lib/Device.cpp

 #include "llvm/ADT/STLExtras.h"
 
 namespace streamexecutor {
 
 Device::Device(PlatformDevice *PDevice) : PDevice(PDevice) {}
 
 Device::~Device() = default;
 
-Expected<std::unique_ptr<Stream>> Device::createStream() {
+Expected<Stream> Device::createStream() {
   Expected<std::unique_ptr<PlatformStreamHandle>> MaybePlatformStream =
       PDevice->createStream();
   if (!MaybePlatformStream) {
     return MaybePlatformStream.takeError();
   }
   assert((*MaybePlatformStream)->getDevice() == PDevice &&
          "an executor created a stream with a different stored executor");
-  return llvm::make_unique<Stream>(std::move(*MaybePlatformStream));
+  return Stream(std::move(*MaybePlatformStream));
 }
 
 } // namespace streamexecutor

parallel-libs/trunk/streamexecutor/lib/Stream.cpp

 ///
 //===----------------------------------------------------------------------===//
 
 #include "streamexecutor/Stream.h"
 
 namespace streamexecutor {
 
 Stream::Stream(std::unique_ptr<PlatformStreamHandle> PStream)
-    : PDevice(PStream->getDevice()), ThePlatformStream(std::move(PStream)) {}
+    : PDevice(PStream->getDevice()), ThePlatformStream(std::move(PStream)),
+      ErrorMessageMutex(llvm::make_unique<llvm::sys::RWMutex>()) {}
 
 Stream::~Stream() = default;
 
 } // namespace streamexecutor