#4673 Make AtWithRho.get() thread-safe

eo-runtime/src/main/java/org/eolang/AtWithRho.java

@@ -5,18 +5,13 @@
 
 package org.eolang;
 
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
+
 /**
  * The attribute that tries to copy object and set \rho to it if it has not already set.
- * This attribute is NOT thread safe!
  *
  * @since 0.36.0
- * @todo #4673:30min The {@link AtWithRho#get()} is not thread safe. If multiple threads
- *  call get() concurrently when the underlying object lacks RHO, each thread will:
- *  1. Pass the !ret.hasRho() check
- *  2. Create its own copy via ret.copy()
- *  3. Attempt to set RHO on its copy
- *  This results in different threads receiving different copies, violating the expectation
- *  that get() returns a consistent view of the attribute's value.
  */
 final class AtWithRho implements Attr {
     /**
@@ -29,6 +24,16 @@ final class AtWithRho implements Attr {
      */
     private final Phi rho;
 
+    /**
+     * Cached result of {@link #get()} to guarantee a consistent view across threads.
+     */
+    private final AtomicReference<Phi> cached;
+
+    /**
+     * Lock guarding the first initialization of {@link #cached}.
+     */
+    private final ReentrantLock lock;
+
     /**
      * Ctor.
      * @param attr Attribute
@@ -37,6 +42,8 @@ final class AtWithRho implements Attr {
     AtWithRho(final Attr attr, final Phi rho) {
         this.original = attr;
         this.rho = rho;
+        this.cached = new AtomicReference<>();
+        this.lock = new ReentrantLock();
     }
 
     @Override
@@ -49,16 +56,29 @@ public Attr copy(final Phi self) {
 
     @Override
     public Phi get() {
-        Phi ret = this.original.get();
-        if (!ret.hasRho()) {
-            ret = ret.copy();
-            ret.put(Phi.RHO, this.rho);
+        Phi ret = this.cached.get();
+        if (ret == null) {
+            this.lock.lock();
+            try {
+                ret = this.cached.get();
+                if (ret == null) {
+                    ret = this.original.get();
+                    if (!ret.hasRho()) {
+                        ret = ret.copy();
+                        ret.put(Phi.RHO, this.rho);
+                    }
+                    this.cached.set(ret);
+                }
+            } finally {
+                this.lock.unlock();
+            }
         }
         return ret;
     }
 
     @Override
     public void put(final Phi phi) {
+        this.cached.set(null);
         this.original.put(phi);
     }
 }

eo-runtime/src/test/java/org/eolang/AtWithRhoTest.java

@@ -4,6 +4,12 @@
  */
 package org.eolang;
 
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import org.hamcrest.MatcherAssert;
 import org.hamcrest.Matchers;
 import org.junit.jupiter.api.Test;
@@ -97,4 +103,44 @@ void copiesWithNewRhoMustCallCopyOnOriginal() {
             Matchers.not(Matchers.is(obj))
         );
     }
+
+    @Test
+    void returnsSameInstanceToConcurrentCallers() throws InterruptedException {
+        final int threads = 16;
+        final Attr attr = new AtWithRho(
+            new AtComposite(new PhDefault(), phi -> phi),
+            new PhDefault()
+        );
+        final Set<Phi> results = ConcurrentHashMap.newKeySet();
+        final CountDownLatch start = new CountDownLatch(1);
+        final CountDownLatch done = new CountDownLatch(threads);
+        final ExecutorService pool = Executors.newFixedThreadPool(threads);
+        try {
+            for (int idx = 0; idx < threads; ++idx) {
+                pool.submit(() -> {
+                    try {
+                        start.await();
+                        results.add(attr.get());
+                    } catch (final InterruptedException ex) {
+                        Thread.currentThread().interrupt();
+                    } finally {
+                        done.countDown();
+                    }
+                });
+            }
+            start.countDown();
+            MatcherAssert.assertThat(
+                "all threads must finish",
+                done.await(10L, TimeUnit.SECONDS),
+                Matchers.is(true)
+            );
+        } finally {
+            pool.shutdownNow();
+        }
+        MatcherAssert.assertThat(
+            "AtWithRho.get() must return the same instance for all concurrent callers",
+            results,
+            Matchers.hasSize(1)
+        );
+    }
 }