[Bug]: 异步批量插入接口 saveInTxAsync 数据量大时 await 无法同步等待-ohos

已关闭

[Bug]: 异步批量插入接口 saveInTxAsync 数据量大时 await 无法同步等待 #301

starfish1987创建于 7月16日关闭于 7月21日

starfish1987

7月16日创建

感谢对OpenHarmony社区的支持与关注，欢迎反馈缺陷。

发生了什么问题？

插入数据量大时（1000 +），多次 await saveInTxAsync(entities) 调用会出现并发执行，无法实现同步

同以下已关闭的issue是同一个问题：
https://gitcode.com/CPF-ApplicationTPC/ohos_dataorm/issues/207

期望行为是什么？

多次 await saveInTxAsync(entities) 调用不应该出现并发，应该同步执行

如何复现该缺陷

多次调用 await saveInTxAsync(entities) 插入 1000+ 数据

库的版本或分支信息

2.3.5

当前使用的DevEco Studio 和 Openharmony/HarmonyOS SDK的版本信息

6.1.1 Relsease

HarmonyOS 6.1.1

其他补充信息

S starfish1987

7月16日添加了label：bug

zhengjinshou成员

7月17日将 longziz 设为负责人

zhengjinshou成员

7月17日评论：

@starfish1987 感谢反馈，我们基于 2.3.5 版本，构造下面3个场景尝试复现问题:
1、多次 await saveInTxAsync 串行调用：3000条/批 x 5批（共15000条）
2、并发触发 saveInTxAsync ：3批 x 3000条（共9000条）同时触发，验证事务隔离和数据完整性
3、循环 await saveInTxAsync：1000条/次 x 10次（共10000条），每轮插入后立即查询

但是上面3个场景都没复现出问题

推测可能与您的具体使用场景有关，能否提供更具体的信息以便于进一步排查:
1、相关代码片段：包括实体类定义、数据库初始化配置、以及业务侧调用 saveInTxAsync 的代码
2、问题的具体表现：您提到的"出现并发执行"，具体是通过什么现象观察到的？（查询数据不一致？事务报错？日志时间线交错？）

starfish1987

7月20日评论：

@starfish1987 感谢反馈，我们基于 2.3.5 版本，构造下面3个场景尝试复现问题:
1、多次 await saveInTxAsync 串行调用：3000条/批 x 5批（共15000条）
2、并发触发 saveInTxAsync ：3批 x 3000条（共9000条）同时触发，验证事务隔离和数据完整性
3、循环 await saveInTxAsync：1000条/次 x 10次（共10000条），每轮插入后立即查询

但是上面3个场景都没复现出问题

推测可能与您的具体使用场景有关，能否提供更具体的信息以便于进一步排查:
1、相关代码片段：包括实体类定义、数据库初始化配置、以及业务侧调用 saveInTxAsync 的代码
2、问题的具体表现：您提到的"出现并发执行"，具体是通过什么现象观察到的？（查询数据不一致？事务报错？日志时间线交错？）

@longziz

通过开启日志观察没有发现报错，但是时间线是有交错，以下是大致代码逻辑：

func(){
Log('delete start')
// 先删除全部数据
await dao.delete()
Log('delete end')
Log('saveInTxAsync start')
// 插入新数据
await dao.saveInTxAsync()
Log('saveInTxAsync end')
}

调用func几次会出现 saveInTxAsync start 和 saveInTxAsync end 之间穿插 delete 和 saveInTxAsync start 日志

如果不易复现可以增加数据量，我们目前2500条数据，在Mate 60 pro 6.1手机上复现概率很高

zhengjinshou成员

7月20日评论：

你好，经验证，在单个函数内多次 await 调用 deleteAllAsync + saveInTxAsync，无论重复多少次，都不会出现问题，例如：

func() {
  await dao.deleteAllAsync()
  await dao.saveInTxAsync(entities)
  await dao.deleteAllAsync()
  await dao.saveInTxAsync(entities)
  // ... 重复
}

说明 await saveInTxAsync 本身能正常同步等待，单次调用和串行调用都没有问题。

问题出现在 func 被并发多次触发的情况（不等待前一个完成就触发下一个）：

// 并发触发
func()  // A
func()  // B（A 还没完成就触发）
func()  // C

此时 A 的 saveInTxAsync 还在执行，B 的 deleteAllAsync 已经开始，会清空 A 刚插入的数据，导致数据不一致。

已尝试在库内部给 *Async 方法加串行化锁，但库层面的串行化只能覆盖单个方法，上面业务场景 func 最小单元是 deleteAllAsync + saveInTxAsync 的组合，库无法感知"这两步属于同一个业务原子单元"。

因此建议在业务侧通过 Promise 链实现串行锁，确保一次只执行一个 func，伪代码如下：

// 业务侧串行化锁
let lockPromise: Promise<void> = Promise.resolve();

const runSerial = (task: () => Promise<void>): Promise<void> => {
  const result = lockPromise.then(task);
  // 即使 task 失败，锁链也要继续，避免后续任务永远阻塞
  lockPromise = result.then(() => {}, () => {});
  return result;
};

// 用法：整个 func 串行执行，避免 delete + save 被拆分
const func = async () => {
  await runSerial(async () => {
    await dao.deleteAllAsync();
    await dao.saveInTxAsync(entities); 
  });
};

// 并发触发也安全
func();  // A
func();  // B（会等 A 完成才执行）
func();  // C（会等 B 完成才执行）

starfish1987

7月20日评论：

你好，经验证，在单个函数内多次 await 调用 deleteAllAsync + saveInTxAsync，无论重复多少次，都不会出现问题，例如：
func() {
 await dao.deleteAllAsync()
 await dao.saveInTxAsync(entities)
 await dao.deleteAllAsync()
 await dao.saveInTxAsync(entities)
 // ... 重复
}
说明 await saveInTxAsync 本身能正常同步等待，单次调用和串行调用都没有问题。

问题出现在 func 被并发多次触发的情况（不等待前一个完成就触发下一个）：
// 并发触发
func()  // A
func()  // B（A 还没完成就触发）
func()  // C
此时 A 的 saveInTxAsync 还在执行，B 的 deleteAllAsync 已经开始，会清空 A 刚插入的数据，导致数据不一致。

已尝试在库内部给 *Async 方法加串行化锁，但库层面的串行化只能覆盖单个方法，上面业务场景 func 最小单元是 deleteAllAsync + saveInTxAsync 的组合，库无法感知"这两步属于同一个业务原子单元"。

因此建议在业务侧通过 Promise 链实现串行锁，确保一次只执行一个 func，伪代码如下：
// 业务侧串行化锁
let lockPromise: Promise<void> = Promise.resolve();

const runSerial = (task: () => Promise<void>): Promise<void> => {
 const result = lockPromise.then(task);
 // 即使 task 失败，锁链也要继续，避免后续任务永远阻塞
 lockPromise = result.then(() => {}, () => {});
 return result;
};

// 用法：整个 func 串行执行，避免 delete + save 被拆分
const func = async () => {
 await runSerial(async () => {
   await dao.deleteAllAsync();
   await dao.saveInTxAsync(entities); 
 });
};

// 并发触发也安全
func();  // A
func();  // B（会等 A 完成才执行）
func();  // C（会等 B 完成才执行）

@longziz

func调用的时候也是 await 的

zhengjinshou成员

7月20日评论：

@starfish1987 感谢反馈，我们基于 2.3.5 版本，构造下面3个场景尝试复现问题:
1、多次 await saveInTxAsync 串行调用：3000条/批 x 5批（共15000条）
2、并发触发 saveInTxAsync ：3批 x 3000条（共9000条）同时触发，验证事务隔离和数据完整性
3、循环 await saveInTxAsync：1000条/次 x 10次（共10000条），每轮插入后立即查询

但是上面3个场景都没复现出问题

推测可能与您的具体使用场景有关，能否提供更具体的信息以便于进一步排查:
1、相关代码片段：包括实体类定义、数据库初始化配置、以及业务侧调用 saveInTxAsync 的代码
2、问题的具体表现：您提到的"出现并发执行"，具体是通过什么现象观察到的？（查询数据不一致？事务报错？日志时间线交错？）

@longziz

通过开启日志观察没有发现报错，但是时间线是有交错，以下是大致代码逻辑：

func(){
Log('delete start')
// 先删除全部数据
await dao.delete()
Log('delete end')
Log('saveInTxAsync start')
// 插入新数据
await dao.saveInTxAsync()
Log('saveInTxAsync end')
}

调用func几次会出现 saveInTxAsync start 和 saveInTxAsync end 之间穿插 delete 和 saveInTxAsync start 日志

如果不易复现可以增加数据量，我们目前2500条数据，在Mate 60 pro 6.1手机上复现概率很高

@starfish1987

上面只提供了 func 内部的使用，需要再提供一下调用 func 的上下文，看下具体调用场景，是同一个函数内串行await？还是在回调/then/Promise.all 中被触发？或者其他调用情况

starfish1987

7月20日评论：

@starfish1987 感谢反馈，我们基于 2.3.5 版本，构造下面3个场景尝试复现问题:
1、多次 await saveInTxAsync 串行调用：3000条/批 x 5批（共15000条）
2、并发触发 saveInTxAsync ：3批 x 3000条（共9000条）同时触发，验证事务隔离和数据完整性
3、循环 await saveInTxAsync：1000条/次 x 10次（共10000条），每轮插入后立即查询

但是上面3个场景都没复现出问题

推测可能与您的具体使用场景有关，能否提供更具体的信息以便于进一步排查:
1、相关代码片段：包括实体类定义、数据库初始化配置、以及业务侧调用 saveInTxAsync 的代码
2、问题的具体表现：您提到的"出现并发执行"，具体是通过什么现象观察到的？（查询数据不一致？事务报错？日志时间线交错？）

@longziz

通过开启日志观察没有发现报错，但是时间线是有交错，以下是大致代码逻辑：

func(){
Log('delete start')
// 先删除全部数据
await dao.delete()
Log('delete end')
Log('saveInTxAsync start')
// 插入新数据
await dao.saveInTxAsync()
Log('saveInTxAsync end')
}

调用func几次会出现 saveInTxAsync start 和 saveInTxAsync end 之间穿插 delete 和 saveInTxAsync start 日志

如果不易复现可以增加数据量，我们目前2500条数据，在Mate 60 pro 6.1手机上复现概率很高

@starfish1987

上面只提供了 func 内部的使用，需要再提供一下调用 func 的上下文，看下具体调用场景，是同一个函数内串行await？还是在回调/then/Promise.all 中被触发？或者其他调用情况

@longziz

客户操作一次（比如数据做了排序），调用一次 await func，如果操作频繁就会出现多次调用

对比使用 dao.saveInTxSync 同步接口同样的场景下目前是不复现的，所以这个接口肯定是有问题提的，https://gitcode.com/CPF-ApplicationTPC/ohos_dataorm/issues/207，这个issue反馈应该也是同样的问题

zhengjinshou成员

7月20日评论：

你好，经验证确实同步方法没有复现，通过代码分析发现，由于历史背景原因，批量插入默认通过遍历依次调用 executeInsert 逐条插入，并不是实际意义的批量插入

可尝试通过下面方式解决:

方式1、使用同步方法： saveInTxSync，内部默认走批量插入 batchInsertSync
方式2、给 dao 设置 setEntityUpdateAble 为 false：
```
+ dao?.setEntityUpdateAble(false); // 可在初始化时调用一次即可

await dao.saveInTxAsync(entities);
```
设置为false之后，会走实际意义的批量插入batchInsert，但需注意这么设置后自增主键不会回写，如果您的实体主键是非自增的（ @Id() 未传 autoincrement: true ），且插入后不需要立即读取 entity 对象的主键值，可以安全使用 setEntityUpdateAble(false)，好处是插入性能更优
方式3、用上面提到的promise串行链的方式

starfish1987

7月20日评论：

你好，经验证确实同步方法没有复现，通过代码分析发现，由于历史背景原因，批量插入默认通过遍历依次调用 executeInsert 逐条插入，并不是实际意义的批量插入

可尝试通过下面方式解决:
方式1、使用同步方法： saveInTxSync，内部默认走批量插入 batchInsertSync
方式2、给 dao 设置 setEntityUpdateAble 为 false：
dao?.setEntityUpdateAble(false); // 可在初始化时调用一次即可

await dao.saveInTxAsync(entities);
设置为false之后，会走实际意义的批量插入batchInsert，但需注意这么设置后自增主键不会回写，如果您的实体主键是非自增的（ @Id() 未传 autoincrement: true ），且插入后不需要立即读取 entity 对象的主键值，可以安全使用 setEntityUpdateAble(false)，好处是插入性能更优
- 方式3、用上面提到的promise串行链的方式

@longziz

嗯嗯，感谢解答，问题我们已经通过改用同步接口解决，但是仍然没有看明白循环单条插入的逻辑有什么问题会导致await不能同步执行：
private async executeInsertInTxWithFor(stmt: DatabaseStatement, entities: Iterable) {

try {
  DaoTraceUtil.startInfo("AbstractDao.insertBatchWithFor", DaoTraceSpace.TraceType.CRUD);
  this.db.beginTransaction();
  if (this.isStandardSQLite) {
    let rawStmt: SQLiteStatement = stmt.getRawStatement();
    for (let entity of entities) {
      this.bindValues(rawStmt, entity)
      let rowId: number = await rawStmt.executeInsert(null, relationalStore.ConflictResolution.ON_CONFLICT_REPLACE);
      this.updateKeyAfterInsertAndAttach(entity, rowId, false);
    }
  } else {
    for (let entity of entities) {
      this.bindValues(stmt, entity)
      let rowId: number = await stmt.executeInsert(null);
      this.updateKeyAfterInsertAndAttach(entity, rowId, false);
    }
  }
  this.db.endTransaction();
} catch (e) {
  DaoLog.e("err_msg:" + e.message, e.stack)
  this.db.rollBack()
} finally {
  DaoTraceUtil.finish("AbstractDao.insertBatchWithFor");
}

}

zhengjinshou成员

7月21日评论：

主要由于 executeInsertInTxWithFor 里面这段遍历：

    for (let entity of entities) {
      this.bindValues(rawStmt, entity)
      let rowId: number = await rawStmt.executeInsert(null, relationalStore.ConflictResolution.ON_CONFLICT_REPLACE);
      this.updateKeyAfterInsertAndAttach(entity, rowId, false);
    }

如果是 2500 条数据，这里会产生 2500 个 await，也就是挂起 2500 次，在事件循环过程中，挂起时会去执行微任务队列里其他任务，2500 次跨度太大了，清空微任务队列时很有可能就会包括了其他 func 的代码

方式一使用同步方法由于不会让出调用栈，所以没有问题

方式二给 dao 设置 setEntityUpdateAble 为 false 本质是让程序执行 batchInsert，这样整个插入是 1 次原子操作，只会产生 1 个 await，其他 func 的代码即使在微任务队列里也要等 batchInsert resolve 后才能执行

starfish1987

7月21日评论：

主要由于 executeInsertInTxWithFor 里面这段遍历：
   for (let entity of entities) {
     this.bindValues(rawStmt, entity)
     let rowId: number = await rawStmt.executeInsert(null, relationalStore.ConflictResolution.ON_CONFLICT_REPLACE);
     this.updateKeyAfterInsertAndAttach(entity, rowId, false);
   }
如果是 2500 条数据，这里会产生 2500 个 await，也就是挂起 2500 次，在事件循环过程中，挂起时会去执行微任务队列里其他任务，2500 次跨度太大了，清空微任务队列时很有可能就会包括了其他 func 的代码

方式一使用同步方法由于不会让出调用栈，所以没有问题

方式二给 dao 设置 setEntityUpdateAble 为 false 本质是让程序执行 batchInsert，这样整个插入是 1 次原子操作，只会产生 1 个 await，其他 func 的代码即使在微任务队列里也要等 batchInsert resolve 后才能执行

@longziz

明白了，感谢解惑

S starfish1987

7月21日关闭了 issue