Web 前端 - 又不仅限于 Web 前端 - 协程锁问题

时间:2021-04-01 本木大人丿人气:0

前言

　　最近两天的 web 前端开发中，早前的锁实现 (自旋锁) 出现了一些不合理的现象，所以有了这片随笔

　　什么是协程锁？能点进这个博客的的你肯定是明白的，不明白的人根本搜不到我这随笔，不多做赘述。

一些个人认识和实现经验

可重入锁：协程由于没有像『线程』那样的变量隔离，即缺少『计数标识』的挂载位置（多线程中计数标识直接或间接挂载在线程对象上），未实现可重入锁之前，编码开发中应该避免嵌套相同锁调用，否则造成死锁，
所以『可重入锁』暂时没有太好的实现思路，希望有大神可以指点迷津。
自旋锁：优点是通俗易懂，一行 while(lock = getLock( lockKey )) await sleep(20) 就可以实现，缺点是不能保证代码的顺序性，造成无法预知的 Bug，如异步打开遮罩 -> 异步关闭遮罩的执行顺序将不可控，弃用。
公平锁：保证 FIFO 策略，使协程对锁先取先得，解决自旋锁的缺陷。
归一锁：类似多线程中的双重检验锁，在多线程中多用于单例的初始化或赋值，在 Web 前端可将重复异步幂等操作消减归一。如『获取用户信息』多个模块都调用此异步函数，只接发出一个『获取用户信息』的请求，多余 1 个的所有协程将进入 Pending 状态，一起等待返回结果。
（结合公平锁，实现顺序等待调用）
等侯锁：只允许一个协程进入操作，多余 1 的所有协程进入 Pending 状态，等待被调用。
（结合公平锁，实现顺序等待调用）

　　暂时想到这么多，欢迎补充。

设计

归一锁：asyncOne( asyncFunction:AsyncFunction, ...keys:any[] )
等侯锁：asyncWait( asyncFunction:AsyncFunction, ...keys:any[] )

实现

锁管理器 ( 核心 )：根据一组 keys ，提供查、加、解锁的能力。
getPendings(keys) / addPendings(keys) / delPendings(keys)

// Synchronized async function
const NEXT = Symbol();
const PREV = Symbol();
type LevelsNode = { [NEXT]?: LevelsMap; pendings?: PendingPromise<any>[]; [PREV]?: LevelsNode; key: string };
type LevelsMap = Map<any, LevelsNode>;

function findSyncNode(syncMap: LevelsMap, keys: any[], creation: false): [LevelsMap | undefined, LevelsNode | undefined, PendingPromise<any>[] | undefined];
function findSyncNode(syncMap: LevelsMap, keys: any[], creation: true): [LevelsMap, LevelsNode, PendingPromise<any>[]];
function findSyncNode(syncMap: LevelsMap, keys: any[], creation: boolean) {
  let prntNode: LevelsNode | undefined;
  let map: LevelsMap | undefined = syncMap;
  for (let i = 0; !!map && i < keys.length - 1; i++) {
    let lastNode = prntNode;
    prntNode = map.get(keys[i]);
    if (!prntNode) {
      if (creation) {
        prntNode = { [NEXT]: new Map(), [PREV]: lastNode, key: keys[i] };
        map.set(keys[i], prntNode);
        map = prntNode[NEXT];
      } else {
        map = undefined;
      }
    } else if (!(map = prntNode[NEXT])) {
      if (creation) {
        prntNode[NEXT] = new Map();
        map = prntNode[NEXT];
      } else {
        if (i < keys.length - 2) prntNode = undefined;
      }
    }
  }
  let mapNode = map?.get(keys[keys.length - 1]);
  if (creation) {
    if (!map) {
      Throwable("Impossible Error: No Prev Map.");
    } else if (!mapNode) {
      map.set(keys[keys.length - 1], (mapNode = { pendings: [], [PREV]: prntNode, key: keys[keys.length - 1] }));
    } else if (!mapNode.pendings) {
      mapNode.pendings = [];
    }
  }
  return [map, mapNode, mapNode?.pendings];
}
const getPendings = (syncMap: LevelsMap, keys: any[]) => findSyncNode(syncMap, keys, false)[2];
const addPendings = (syncMap: LevelsMap, keys: any[]) => findSyncNode(syncMap, keys, true)[2];
const delPendings = (syncMap: LevelsMap, keys: any[]) => {
  const [finalMap, finalVal] = findSyncNode(syncMap, keys, false);
  if (!!finalMap && !!finalVal) {
    // delete pending
    delete finalVal.pendings;
    // delete above including self
    tryDeleteNodeAndAbove(syncMap, finalVal);
  }
};
const tryDeleteNodeAndAbove = (syncMap: LevelsMap, node?: LevelsNode) => {
  while (!!node) {
    const nextMap = node[NEXT];
    if (!node.pendings && (!nextMap || nextMap.size === 0)) {
      const nodeKey = node.key;
      node = node[PREV];
      const map = node?.[NEXT] || syncMap;
      map.delete(nodeKey);
    } else {
      break;
    }
  }
};

归一锁算法：

const asyncOneMap: LevelsMap = new Map<any, LevelsNode>();
export const asyncOne = async <T>(call: () => Promise<T>, ...keys: any[]): Promise<T> => {
  let pendings = getPendings(asyncOneMap, keys);
  if (!!pendings) {
    return (pendings[pendings.length] = pendingResolve<T>());
  } else {
    pendings = addPendings(asyncOneMap, keys);
    try {
      const result = await call.call(null);
      pendings.forEach(p => setTimeout(() => p.resolve(result)));
      return result;
    } catch (e) {
      pendings.forEach(p => setTimeout(() => p.reject(e)));
      throw e;
    } finally {
      delPendings(asyncOneMap, keys);
    }
  }
};

等侯锁算法：

const asyncWaitMap: LevelsMap = new Map<any, LevelsNode>();
export const asyncWait = async <T>(call: () => Promise<T>, ...keys: any[]) => {
  let pendings = getPendings(asyncWaitMap, keys);
  /*sleep*/ if (!!pendings) await (pendings[pendings.length] = pendingResolve<void>());
  /*continue-------------*/ else pendings = addPendings(asyncWaitMap, keys);
  try {
    return await call.call(null);
  } finally {
    const next4awaken = pendings.shift();
    /*awaken the next*/ if (next4awaken !== undefined) next4awaken.resolve();
    /*unlock-----------------------------------*/ else delPendings(asyncWaitMap, keys);
  }
};

依赖项 ( 第二关键 )：pendingResolve<T>: <T>()=>PendingPromise<T>
返回一个永久 pending 状态的 Promise, 充当协程断点的角色，必要时才手动 resolve / reject。
本文不多赘述，请参考我的另一篇随笔: Web 前端 - 浅谈外部手动控制 Promise 状态：PendingPromise<T>
依赖项 ( 轻微重要 )：aw: ( ms: number)=>Promise<void>
类似于多线程语言中的 sleep( ms:number )
本文不多赘述，请参考我的另一篇随笔: Web 前端 - 优雅地 Callback 转 Promise ：aw

加载全部内容