parallel_select_and_update.js

This example is similar to nodejs/sequelize/raw/parallel_update_async.js, but now we are doing a separate SELECT, later followed by an update:

SELECT FROM to get i
update on Js code newI = i + 1
UPDATE SET the newI

Although this specific example is useless in itself, as we could just use UPDATE "MyInt" SET i = i + 1 as in nodejs/sequelize/raw/parallel_update_async.js, which automatically solves any concurrency issue, this kind of code could be required for example if the update was a complex function not suitably implemented in SQL, or if the update depends on some external data source.

Sample execution:

node --unhandled-rejections=strict ./parallel_select_and_update.js p 2 10 'READ COMMITTED'

which does:

PostgreSQL, see other databases options at SQL example
2 threads
10 increments on each thread

Another one:

node --unhandled-rejections=strict ./parallel_select_and_update.js p 2 10 'READ COMMITTED' 'FOR UPDATE'

this will run SELECT FOR UPDATE rather than just SELECT

Observed behaviour under different SQL transaction isolation levels:

READ COMMITTED: fails. Nothing in this case prevents:
- thread 1: SELECT, obtains i = 0
- thread 2: SELECT, obtains i = 0
- thread 2: newI = 1
- thread 2: UPDATE i = 1
- thread 1: newI = 1
- thread 1: UPDATE i = 1
REPEATABLE READ: works. the manual mentions that if multiple concurrent updates would happen, only the first commit succeeds, and the following ones fail and rollback and retry, therefore preventing the loss of an update.
READ COMMITTED + SELECT FOR UPDATE: works. And does not do rollbacks, which probably makes it faster. With p 10 100, REPEATABLE READ was about 4.2s and READ COMMITTED + SELECT FOR UPDATE 3.2s on Lenovo ThinkPad P51 (2017).
SELECT FOR UPDATE should be enough as mentioned at: www.postgresql.org/docs/13/explicit-locking.html#LOCKING-ROWS
FOR UPDATE causes the rows retrieved by the SELECT statement to be locked as though for update. This prevents them from being locked, modified or deleted by other transactions until the current transaction ends. That is, other transactions that attempt UPDATE, DELETE, SELECT FOR UPDATE, SELECT FOR NO KEY UPDATE, SELECT FOR SHARE or SELECT FOR KEY SHARE of these rows will be blocked until the current transaction ends; conversely, SELECT FOR UPDATE will wait for a concurrent transaction that has run any of those commands on the same row, and will then lock and return the updated row (or no row, if the row was deleted). Within a REPEATABLE READ or SERIALIZABLE transaction, however, an error will be thrown if a row to be locked has changed since the transaction started. For further discussion see Section 13.4.

A non-raw version of this example can be seen at: nodejs/sequelize/parallel_select_and_update.js.

nodejs/sequelize/raw/parallel_select_and_update.js

#!/usr/bin/env node

// https://cirosantilli.com/file/sequelize/raw/parallel_select_and_update.js

const assert = require('assert');
const common = require('../common');

async function inc(sequelize, n, isolation, for_update) {
  for (let i = 0; i < n; i++) {
    await common.transaction(sequelize, isolation, async sequelize => {
      ;[rows, meta] = await sequelize.query(`SELECT * FROM "MyInt" ${for_update}`)
      const newI = rows[0].i + 1
      await sequelize.query(`UPDATE "MyInt" SET i = ${newI}`)
    })
  }
}

// CLI arguments.
let nthreads;
if (process.argv.length > 3) {
  nthreads = parseInt(process.argv[3], 10)
} else {
  nthreads = 2;
}
let n;
if (process.argv.length > 4) {
  n = parseInt(process.argv[4], 10)
} else {
  n = 10;
}
const isolation = process.argv[5]
let for_update = process.argv[6]
if (for_update === undefined) {
  for_update = ''
}

const sequelizes = common.sequelizes(nthreads, __filename, process.argv[2])
const sequelize = sequelizes[0]
;(async () => {
await common.drop(sequelize, 'MyInt')
await sequelize.query(`CREATE TABLE "MyInt" ( i INTEGER NOT NULL )`)
await sequelize.query(`INSERT INTO "MyInt" VALUES (0)`)
const arr = []
for (let i = 0; i < nthreads; i++) {
  arr.push(inc(sequelizes[i], n, isolation, for_update))
}
await Promise.all(arr)
let [rows, meta] = await sequelize.query(`SELECT * FROM "MyInt"`)
assert.strictEqual(rows[0].i, nthreads * n)
})().finally(() => {
  return Promise.all(sequelizes.map(s => s.close()))
});

nodejs/sequelize/raw/parallel_select_and_update.js

 Ancestors

 Incoming links