Repository: eclipse-vertx/vert.x

Problem statement:
Race condition in AsyncFileImpl writesOutstanding
If an AsyncFileImpl is written to from a thread different to the context that created then the long counter writesOutstanding can get into an invalid state. This is because the counter is updated from both the caller's thread and also on the context when a write is complete.

Fix is to use an AtomicLong for writesOutstanding:

https://gist.github.com/purplefox/723fa69a1330fb4f4f74d4a5dd5e6658


Hints:
a couple of questions:

- can you tell the unsynchronized write that happens ? is it the handler passed to the doWrite method that seems unsynchronized when called back
- is the atomicity of the long necessary ?
- could it be fixed using synchronized instead or setting the field as volatile ?
```
          context.runOnContext((v) -> {
            writesOutstanding -= buff.limit();
            handler.handle(Future.succeededFuture());
          });
```

I think the issue is that writeOutstanding can be increased here:

https://github.com/eclipse/vert.x/blob/master/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java#L375

At the same time it is decreased here:

https://github.com/eclipse/vert.x/blob/master/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java#L395

If write is called from a different thread to the context thread.

I tried using synchronized to fix initially but this resulted in a deadlock. Volatile can't be used as volatile adds aren't atomic.
can you provide a reproducer ?
Sure, will try and put one together :)
Here's a reproducer that can run inside vertx-core FileSystemTest:

https://gist.github.com/purplefox/6de14775d331395ceeb01168367715f4

It needs to be run with many iterations to reproduce (hence the @Repeat)
If you run the test enough times you will find it hangs eventually, this is because the close() method never completes. It doesn't complete because it is waiting for an invalid value of outstanding writes.
makes sense thank!

Patch:
diff --git a/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java b/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java
--- a/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java
+++ b/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java
@@ -148,11 +148,15 @@ private synchronized AsyncFile doWrite(Buffer buffer, long position, Handler<Asy
     Handler<AsyncResult<Void>> wrapped = ar -> {
       if (ar.succeeded()) {
         checkContext();
-        if (writesOutstanding == 0 && closedDeferred != null) {
-          closedDeferred.run();
-        } else {
-          checkDrained();
+        Runnable action;
+        synchronized (AsyncFileImpl.this) {
+          if (writesOutstanding == 0 && closedDeferred != null) {
+            action = closedDeferred;
+          } else {
+            action = this::checkDrained;
+          }
         }
+        action.run();
         if (handler != null) {
           handler.handle(ar);
         }
@@ -372,7 +376,9 @@ private void doWrite(ByteBuffer buff, long position, long toWrite, Handler<Async
     if (toWrite == 0) {
       throw new IllegalStateException("Cannot save zero bytes");
     }
-    writesOutstanding += toWrite;
+    synchronized (this) {
+      writesOutstanding += toWrite;
+    }
     writeInternal(buff, position, handler);
   }
 
@@ -392,7 +398,9 @@ public void completed(Integer bytesWritten, Object attachment) {
         } else {
           // It's been fully written
           context.runOnContext((v) -> {
-            writesOutstanding -= buff.limit();
+            synchronized (AsyncFileImpl.this) {
+              writesOutstanding -= buff.limit();
+            }
             handler.handle(Future.succeededFuture());
           });
         }


Test patch:
diff --git a/src/test/java/io/vertx/test/core/FileSystemTest.java b/src/test/java/io/vertx/test/core/FileSystemTest.java
--- a/src/test/java/io/vertx/test/core/FileSystemTest.java
+++ b/src/test/java/io/vertx/test/core/FileSystemTest.java
@@ -53,8 +53,10 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static io.vertx.test.core.TestUtils.*;
 
@@ -1679,4 +1681,35 @@ private void deleteFile(String fileName) {
     File file = new File(testDir + pathSep + fileName);
     file.delete();
   }
-}
+
+  // @Repeat(times=1000)
+  @Test
+  public void testAsyncFileConcurrency() throws Exception {
+    String fileName = "some-file.dat";
+
+    AtomicReference<AsyncFile> arFile = new AtomicReference<>();
+    CountDownLatch latch = new CountDownLatch(1);
+    vertx.fileSystem().open(testDir + pathSep + fileName, new OpenOptions(), ar -> {
+      if (ar.succeeded()) {
+        AsyncFile af = ar.result();
+        arFile.set(af);
+      } else {
+        fail(ar.cause().getMessage());
+      }
+      latch.countDown();
+    });
+    awaitLatch(latch);
+
+    AsyncFile af = arFile.get();
+
+    Buffer buff = Buffer.buffer(randomByteArray(4096));
+    for (int i = 0; i < 100000; i++) {
+      af.write(buff);
+    }
+
+    af.close(onSuccess(v -> {
+      testComplete();
+    }));
+
+    await();
+  }}