Java Concurrency #3 - Locking

在前篇我們透過 Atomic Variable 成功替 Servlet 加了一個 state，但如果我有更多的 state 要維護，是不是就是多加一些 Atomic state 就好了？

• 假設有一個情境，我們要在 Servlet 針對某個計算結果作 caching，以便在另一個 request 進來的時候不用重新計算，我們可能會如以下實作:

    public class UnsafeCachingFactorizer implements Servlet {
        private final AtomicReference<BigInteger> lastNumber = new AtomicReference<BigInteger>();
        private final AtomicReference<BigInteger[]> lastFactors = new AtomicReference<BigInteger[]>();
  
        public void service(ServletRequest req, ServletResponse resp) {
            BigInteger i = extractFromRequest(req);
            if (i.equals(lastNumber.get()))
                encodeIntoResponse(resp, lastFactors.get());
            else {
                BigInteger[] factors = factor(i);
                lastNumber.set(i);
                lastFactors.set(factors);
                encodeIntoResponse(resp, factors);
            }
        }
    }

  • 如果你以為這個是一個 Thread safe class，那就錯了
  • 雖然 lastNumber 與 lastFactors 個別都是 Thread safe state，但 11 與 12 行卻是類似 Compound Operation，兩行指令必須同時發生，才會是 Thread safe
  • 同樣的，如果今天有一個方法是要取得它們的值，也會因為 11-12 行非 Atomic 而取到錯誤的結果

• 如果有多個相關的 state，要如何保持它們的 Thread safety 呢？

  • 一種方法是透過 Java 的 Intrinsic lock，也就是 synchronized 這個保留字所建立的區塊，而建立這個區塊有兩個部份

    • 一個能當 lock 的物件
    • 一個被 lock 所保護的區塊

  • 在 Java 中的 Intrinsic lock 是一種 mutex，換句話說同時最多只有一個執行緒可以持有 lock

    • 正因為這個特性，取得 lock 的執行緒在執行區塊時，就等同是 Atomic operation
    • 也就是多個指令被視為一個單一、不可分割的單位，Atomic 的定義不正是如此？

  • 因此，上面的 Servlet 我們可以調整如下:

      public class SynchronizedFactorizer implements Servlet {
          private BigInteger lastNumber;
          private BigInteger[] lastFactors;
    
          public synchronized void service(ServletRequest req, ServletResponse resp) {
              BigInteger i = extractFromRequest(req);
              if (i.equals(lastNumber))
                  encodeIntoResponse(resp, lastFactors);
              else {
                  BigInteger[] factors = factor(i);
                  lastNumber = i;
                  lastFactors = factors;
                  encodeIntoResponse(resp, factors);
              }
          }
      }

    • 如此一來整個方法在同一時間只會有一個 Thread 可存取，進而確保了 11-12 行的操作是 atomic，使得類別本身變成執行緒安全(Thread safety)
    • 但這樣的做法會大大影響效能，因為整個方法都被鎖住了(後面會再提到如何解)，而且會造成 liveness 問題(所有的執行緒都在等著取得 lock)

  • Instrinsic lock 還有一個特性就是 Reentrancy

    • 當一個執行緒取得 lock 的時候，另一個執行緒會被 block
    • 那如果已經取得 lock 的執行緒第二次進到這個方法呢？難道會被它自己 block 住造成 deadlock？
    • 答案是否定的，因為 Reentrancy 的特性，如果該執行緒已持有 lock，那它就不會有進入障礙
    • 這部份的實作是靠 JVM 在執行緒取得 lock 時作持有者的記錄，並針對那個 lock 的取得次數加 1
    • 當持有者退出 synchronized block，取得次數就會減1，而當取得次數歸零時，lock 就會被釋放，其他的 thread 就能嘗試取得 lock

  • 只要加了 syncrhonized 就一定執行緒安全了嗎？

    • 不，多執行緒會存取某個 state 的話，每一個存取它的地方都要 synchronize
    • 此外，用 lock 來保護某個區塊的時候，也必須是同一個 lock
    • 常見的誤會就是只有寫入共用的 state 時才需要 synchronize
    • 小結: 每一個會被多執行緒共用的 state，都必須被唯一的 lock 保護著

  • 一個常見的作法就是把所有 mutable 的 state 都放到一個類別，並處理並行的存取，以確保這些 state 都是執行緒安全的

    • 比方說 java.util.Vector，或是其他的 synchronized collection(e.g. ConcurrentHashMap, CopyOnWriteArrayList...等)
    • 但要特別留意如果加新方法的時候也要記得處理同步的部份

  • 並非所有的 state 都要處理同步，只有 mutable 的 state 需要

    • 假設有一個單執行緒的程式，其內容是針對大量的資料作處理，單執行緒是不需要處理同步機制的
    • 如果有一天為了防止意外發生，定義了一個 TimerTask，並定期針對上面的資料處理作進度的快照(Snapshot)，此時意外就會發生了, 因為不再只有一個執行緒存取到資料，必須處理同步機制才行
    • 這時候就要看程式邏輯中的 invariant 是牽涉到一個 state 還是多個 state，如果有多個 state，那全部都必須在取得 lock 的情況下去異動它
    • 另外，有加 synchronized 的方法的確可以確保單一操作 Thread safety，但如果有兩個以上的 synchronized 方法，額外的機制就必須被加上了(後面的篇章會再提幾個技巧)

  • 不要無腦將整個方法都用 synchronized 包住

    • 如果有耗時的操作如網路連線、檔案I/O等…盡可能將它們排除在外

    • 不這麼做的話，很容易在大流量的情況下卡住所有的 request

    • 之前的 SynchronizedFactorizer 可以調整如下:

        public class SynchronizedFactorizer implements Servlet {
            private BigInteger lastNumber;
            private BigInteger[] lastFactors;
            private long hits;
            private long cacheHits;
      
            public synchronized long getHits() {
                return hits;
            }
      
            public synchronized double getCacheHitRatio() {
                return (double) cacheHits / (double) hits;
            }
      
            public void service(ServletRequest req, ServletResponse resp) {
                BigInteger i = extractFromRequest(req);
                BigInteger[] factors = null;
                synchronized (this) {
                    ++hits;
                    if (i.equals(lastNumber)) {
                        ++cacheHits;
                        factors = lastFactors.clone();
                    }
                }
                if (factors == null) {
                    factors = factor(i);
                    synchronized (this) {
                        lastNumber = i;
                        lastFactors = factors.clone();
                    }
                }
                encodeIntoResponse(resp, factors);
            }
        }

      • 如上所示，service 這個方法不再整個用 synchronized 包了
      • 相對的，是在有 shared state 有讀寫的地方才這麼做
      • 這樣真正耗效能且只用到 local stack 的地方(計算階乘數)就不會因為要等 lock 而卡很久 => 在 request 量大的時候全部人等一個 thread 算完
      • 大原則就是盡量讓 synchronized 的區塊小，但不要太極端(e.g. ++hit 也包在 synchronized 區塊)，因為取得跟釋放 lock 都是有點耗效能的