Resolve deadlocks in FileSystemContext by removing 'synchronized'

core/client/fs/src/main/java/alluxio/client/file/FileSystemContext.java

@@ -63,6 +63,9 @@
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
@@ -158,9 +161,11 @@ public class FileSystemContext implements Closeable {
   /** Whether to do URI scheme validation for file systems using this context.  */
   private boolean mUriValidationEnabled = true;
 
+  private final Lock mWorkerInfoListLock = new ReentrantLock();
+
   /** Cached map for workers. */
-  @GuardedBy("this")
-  private volatile List<BlockWorkerInfo> mWorkerInfoList = null;
+  @GuardedBy("mWorkerInfoList")
+  private final AtomicReference<List<BlockWorkerInfo>> mWorkerInfoList = new AtomicReference<>();
 
   /** The policy to refresh workers list. */
   @GuardedBy("this")
@@ -636,11 +641,14 @@ public synchronized WorkerNetAddress getNodeLocalWorker() throws IOException {
    *
    * @return the info of all block workers eligible for reads and writes
    */
-  public synchronized List<BlockWorkerInfo> getCachedWorkers() throws IOException {
-    if (mWorkerInfoList == null || mWorkerInfoList.isEmpty() || mWorkerRefreshPolicy.attempt()) {
-      mWorkerInfoList = getAllWorkers();
+  public List<BlockWorkerInfo> getCachedWorkers() throws IOException {
+    synchronized (mWorkerInfoList) {
+      if (mWorkerInfoList.get() == null || mWorkerInfoList.get().isEmpty()
+          || mWorkerRefreshPolicy.attempt()) {
+        mWorkerInfoList.set(getAllWorkers());
+      }
+      return mWorkerInfoList.get();
     }
-    return mWorkerInfoList;
   }
 
   /**