Pipelines

ทดสอบบน Redis 8.10.2 (redis:8.10.2-alpine), redis-cli และ redis-benchmark 8.10.2, ioredis 6.0.0, Node 24 บน Mac (Apple silicon) เวลาทั้งหมดมาจากการรันของผมผ่าน port forward ของ Docker (สคริปต์) หรือผ่าน loopback ในคอนเทนเนอร์ (redis-benchmark) ค่าเหล่านี้แสดงรูปร่างของผล ไม่ใช่ตัวเลขของคุณ

Pipelining คืออะไร?

ทุกคำสั่งมีค่า round trip: client ส่ง รอคำตอบ แล้วจึงส่งคำสั่งถัดไป บนเครือข่ายที่ latency ไม่มากก็ตาม การรอนั้นครองเวลาทั้งหมด: เซิร์ฟเวอร์ทำได้ 100k คำสั่งต่อวินาที แต่ client ที่รอแต่ละคำตอบส่งได้มากสุด 1 / RTT (Redis pipelining) Pipelining ส่งหลายคำสั่งใน write เดียวโดยไม่รอ เซิร์ฟเวอร์รันตามลำดับและคืนคำตอบพร้อมกัน มันเป็นการเพิ่มประสิทธิภาพเครือข่าย ไม่ใช่ฟีเจอร์ด้าน atomicity

Pipeline ใน ioredis

// @run
import { Redis } from 'ioredis';

const url = process.env.REDIS_URL!;
const redis = new Redis(url);
const N = 5000;
const ms = (t: number) => `${(performance.now() - t).toFixed(0)} ms`;

// 1) N round trips vs one pipeline vs ioredis auto-pipelining
let t = performance.now();
for (let i = 0; i < N; i++) await redis.set(`b3:p:${i}`, 'x');
console.log(`${N} awaited SETs, one round trip each : ${ms(t)}`);

t = performance.now();
const pipe = redis.pipeline();
for (let i = 0; i < N; i++) pipe.set(`b3:p:${i}`, 'x');
await pipe.exec();
console.log(`${N} SETs in one pipeline              : ${ms(t)}`);

const auto = new Redis(url, { enableAutoPipelining: true });
t = performance.now();
await Promise.all(Array.from({ length: N }, (_, i) => auto.set(`b3:p:${i}`, 'x')));
console.log(`${N} SETs, Promise.all + autoPipelining: ${ms(t)}`);
auto.disconnect();

// 2) batch size: 50 000 commands sent in batches of different sizes
for (const size of [1, 10, 100, 1000, 10000]) {
  const total = size === 1 ? 5000 : 50000; // size 1 = no batching, keep it short
  t = performance.now();
  for (let i = 0; i < total; i += size) {
    const p = redis.pipeline();
    for (let j = i; j < Math.min(i + size, total); j++) p.set(`b3:b:${j}`, 'x');
    await p.exec();
  }
  console.log(`batch ${String(size).padStart(5)}: ${((total / (performance.now() - t)) * 1000).toFixed(0)} commands/s`);
}

// 3) exec() does NOT reject when a command inside fails: you get [error, result] tuples
await redis.set('b3:str', 'text');
const res = await redis.pipeline().set('b3:a', '1').incr('b3:str').get('b3:a').exec();
console.log('pipeline result:', JSON.stringify(res!.map(([e, r]) => [e ? e.message : null, r])));

// the helper an assistant forgets to write: fail loudly on the first per-command error
async function execOrThrow(p: ReturnType<Redis['pipeline']>) {
  const out = await p.exec();
  const failed = out!.findIndex(([err]) => err);
  if (failed !== -1) throw new Error(`command #${failed} failed: ${out![failed]![0]!.message}`);
  return out!.map(([, r]) => r);
}
try {
  await execOrThrow(redis.pipeline().set('b3:a', '2').incr('b3:str'));
} catch (e) {
  console.log('execOrThrow ->', (e as Error).message);
}

// 4) a pipeline is not atomic; wrap it in MULTI when you need that
const tx = await redis.multi().incr('b3:n').incr('b3:n').exec();
console.log('MULTI inside the same round trip:', JSON.stringify(tx));

const del = [...Array.from({ length: 50000 }, (_, i) => `b3:b:${i}`), ...Array.from({ length: N }, (_, i) => `b3:p:${i}`), 'b3:str', 'b3:a', 'b3:n'];
for (let i = 0; i < del.length; i += 1000) await redis.unlink(del.slice(i, i + 1000));
await redis.quit();

ผลลัพธ์จริง (Mac เครื่องนี้):

5000 awaited SETs, one round trip each : 1688 ms
5000 SETs in one pipeline              : 17 ms
5000 SETs, Promise.all + autoPipelining: 32 ms
batch     1: 3078 commands/s
batch    10: 27342 commands/s
batch   100: 182387 commands/s
batch  1000: 406093 commands/s
batch 10000: 472466 commands/s
pipeline result: [[null,"OK"],["ERR value is not an integer or out of range",null],[null,"1"]]
execOrThrow -> command #1 failed: ERR value is not an integer or out of range
MULTI inside the same round trip: [[null,1],[null,2]]

สิ่งที่พิสูจน์ได้:

round trip คือต้นทุน SET แบบ await 5000 ครั้งใช้ 1688 ms (ราว 0.34 ms ต่อครั้ง ส่วนใหญ่คือ port forward ของ Docker); 5000 ครั้งเดียวกันใน pipeline เดียวใช้ 17 ms เร็วกว่าราว 100 เท่า
enableAutoPipelining ทำให้ ioredis รวมคำสั่งที่ออกใน tick เดียวกัน: Promise.all บน set ธรรมดา 5000 ครั้งใช้ 32 ms โดยไม่ต้องสร้าง pipeline เอง
ขนาด batch ให้ผลตอบแทนลดลง throughput เพิ่มจาก 3k คำสั่ง/วินาที (ไม่ batch) เป็น 406k ที่ 1000 และ 472k ที่ 10 000 เอกสาร Redis แนะนำ batch ราว 10k เพราะเซิร์ฟเวอร์เก็บคำตอบทั้งหมดในหน่วยความจำจนกว่าจะถูกอ่าน ความเร็วเกือบเท่าเดิมและหน่วยความจำมีขอบเขต
exec() ไม่ reject เมื่อคำสั่งข้างในล้มเหลว มัน resolve ด้วยอาร์เรย์ของ tuple [error, result] INCR บนค่าข้อความให้ error tuple ที่ตำแหน่ง 1 ขณะที่ SET ก่อนหน้าและ GET หลังจากนั้นสำเร็จ ถ้าคุณ await pipeline.exec() ใน try/catch เฉยๆ จะไม่มีอะไรถูกจับเลย execOrThrow แสดง loop ที่คุณต้องเขียนเอง
pipeline ไม่ใช่ transaction บรรทัดสุดท้ายแสดง multi()...exec() (MULTI/EXEC ที่ส่งใน round trip เดียว): ได้ทั้ง round trip น้อยลง และ atomicity ผลลัพธ์มีรูปร่าง tuple เดียวกัน

`redis-cli --pipe` สำหรับโหลดข้อมูลจำนวนมาก

redis-cli --pipe อ่านคำสั่งจาก stdin และสตรีมไปยังเซิร์ฟเวอร์โดยไม่รอคำตอบ รับ RESP และคำสั่งแบบ inline (หนึ่งคำสั่งต่อบรรทัด) ก็ใช้ได้:

# RESP: *<argc>, then $<len> and the bytes for each argument
printf '*3\r\n$3\r\nSET\r\n$2\r\nk1\r\n$5\r\nhello\r\n*3\r\n$3\r\nSET\r\n$2\r\nk2\r\n$5\r\nworld\r\n' | redis-cli --pipe

# inline commands, one per line (a file works the same way: cat commands.txt | redis-cli --pipe)
printf 'SET k1 hello\nBOGUS x\nSET k3 1\n' | redis-cli --pipe

ผลลัพธ์จริงของสองการเรียก:

All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 2

All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 1, replies: 3
ERR unknown command 'BOGUS', with args beginning with: 'x'

บรรทัดสรุปนับ reply และ error และพิมพ์ error แต่ละตัว คำสั่งอื่นยังรันอยู่: เหมือน ioredis --pipe ไม่หยุดที่ความล้มเหลวแรก

วัดด้วย redis-benchmark ในคอนเทนเนอร์ (loopback, -n 100000 -t set):

redis-benchmark -q -n 100000 -t set -P 1     # one command per round trip
redis-benchmark -q -n 100000 -t set -P 16    # 16 commands per round trip
redis-benchmark -q -n 100000 -t set -P 100

SET: 211864.41 requests per second, p50=0.119 msec        (-P 1)
SET: 1351351.38 requests per second, p50=0.511 msec       (-P 16)
SET: 1694915.25 requests per second, p50=2.783 msec       (-P 100)

16 คำสั่งต่อ round trip ให้ throughput ราว 6 เท่าแม้บน loopback และ latency ต่อคำขอเพิ่มขึ้น (p50 0.12 เป็น 0.51 ms) เพราะแต่ละคำตอบต้องรอ batch ของมัน บนเครือข่ายจริงที่ RTT มากกว่า loopback มาก ช่องว่างยิ่งกว้าง

Pipeline เทียบกับ transaction

ฟีเจอร์	Pipeline	Transaction (`MULTI`/`EXEC`)
Round trip	หนึ่ง (หรือไม่กี่ครั้ง)	หนึ่งถ้าส่งพร้อมกัน ไม่เช่นนั้นหนึ่งต่อคำสั่ง
Atomicity	ไม่ใช่: client อื่นแทรกได้	ใช่: ไม่มีอะไรขัด
Rollback	ไม่มี	ไม่มี
Error ต่อคำสั่ง	คืนในลิสต์คำตอบ	คืนในลิสต์คำตอบ (error ตอน queue ยกเลิกทั้งหมด ดู Transactions)
กรณีใช้	Throughput / โหลดจำนวนมาก	เขียนเป็นกลุ่มอย่างปลอดภัย

pipeline คือ การเพิ่มประสิทธิภาพเครือข่าย ส่วน transaction คือ การรับประกัน atomicity รวมกันได้โดยส่ง MULTI/EXEC ใน pipeline ดังที่ multi() ใน ioredis ทำ

SET a 1
SET b 2
SET c 3
MGET a b c

Reading AI-generated code

AI เขียนตัวนำเข้าข้อมูลจำนวนมากและห่อด้วย try/catch “เพื่อจัดการความล้มเหลว”:

🐛 หา bug ในโค้ดนี้

async function importUsers(rows: Array<{ id: string; name: string }>) {
try {
  const pipeline = redis.pipeline();
  for (const r of rows) pipeline.hset(`user:${r.id}`, 'name', r.name);
  await pipeline.exec();
  console.log(`imported ${rows.length} users`);
} catch (err) {
  console.error('import failed', err);
}
}

เฉลย

pipeline.exec() resolve แม้คำสั่งจะล้มเหลว ในการรัน INCR บนค่าข้อความให้ error ภายใน อาร์เรย์ผลลัพธ์และไม่มีอะไรถูกโยน ดังนั้น WRONGTYPE บน key เดียว (เช่น user id ที่มีอยู่แล้วเป็นสตริง) จึงถูกรายงานว่า “imported N users” catch ทำงานเฉพาะปัญหาระดับ connection ตรวจทุก tuple:

const results = (await pipeline.exec())!;
const failed = results.filter(([err]) => err);
if (failed.length) throw new Error(`${failed.length} of ${rows.length} commands failed: ${failed[0]![0]!.message}`);

Mental Model

pipeline คือ การนำจดหมายเป็นปึกไปหย่อนตู้ไปรษณีย์ในเที่ยวเดียว แทนที่จะเดินไปทีละฉบับ พนักงานยังอ่านทีละฉบับตามลำดับ และลูกค้าคนอื่นถูกเรียกใช้บริการคั่นระหว่างสองฉบับของคุณได้ คำตอบกลับมาเป็นมัดเดียว และการที่ฉบับหนึ่งถูกปฏิเสธไม่หยุดฉบับที่เหลือ

Trade-offs

ตัวเลือก	ข้อดี	ข้อเสีย
Pipeline / `autoPipelining`	round trip น้อยลง 100 เท่าในการรัน (17 ms เทียบ 1688 ms)	ไม่ atomic; error ซ่อนในอาร์เรย์คำตอบ
Batch ราว 1000-10 000	throughput ใกล้จุดสูงสุด หน่วยความจำคำตอบมีขอบเขต	ต้องแบ่งงานและรอแต่ละ batch
Pipeline ยักษ์อันเดียว	round trip น้อยที่สุด	เซิร์ฟเวอร์และ client บัฟเฟอร์ทุกคำตอบในหน่วยความจำ
`redis-cli --pipe`	โหลดจำนวนมากจากไฟล์หรือโปรแกรมอื่นได้เร็ว	error ถูกพิมพ์แต่ไม่หยุดการโหลด; ต้องสร้าง input ที่ถูกต้องเอง
`MULTI`/`EXEC` ใน pipeline	batch แบบ atomic ในเที่ยวเดียว	batch รันเป็นบล็อกเดียวที่ไม่ถูกขัด; ให้เล็กไว้

Common Misconceptions

“pipeline เป็น atomic” client อื่นรันแทรกระหว่างคำสั่งของมันได้
“try/catch รอบ exec() จับคำสั่งที่ล้มเหลว” ใน ioredis ไม่จับ ตรวจ tuple
“batch ใหญ่กว่าเร็วกว่าเสมอ” ผลตอบแทนแบนราบ (406k เทียบ 472k คำสั่ง/วินาทีจาก 1000 เป็น 10 000) ขณะที่หน่วยความจำคำตอบโตขึ้น
“pipelining ลด latency” มันเพิ่ม throughput; แต่ละคำตอบอาจรอนานขึ้น (-P 16 p50 0.51 ms เทียบ 0.12 ms)

Common mistakes

เข้าใจว่า pipeline เป็น atomic — ห่อด้วย MULTI/EXEC ถ้าต้อง atomic
ไม่สนใจ result tuple — ความล้มเหลวผ่านไปเงียบๆ (พิสูจน์ข้างบน)
pipeline เดียวขนาดมหึมา — ส่งเป็น batch ราว 10k หรือน้อยกว่า
pipeline คำสั่งที่พึ่งคำตอบของกันและกัน — ทั้งหมดถูกส่งก่อนคำตอบใดจะมาถึง ใช้ Lua หรือ MULTI กับ WATCH สำหรับ read-then-write
ป้อน --pipe ด้วย input ที่ไม่ตรวจสอบ — บรรทัดที่เสียกลายเป็น error กลางการโหลด ดูบรรทัดสรุป

Real-world example

เอกสารของ Redis เอง วัดผลและให้คำแนะนำ batch 10k (Redis pipelining) และเปรียบเทียบ pipelining กับ scripting ด้วย: สคริปต์ดีกว่าเมื่อคำสั่งต้องใช้คำตอบก่อนหน้า เพราะคำสั่งที่ pipeline อ่านผลก่อนหน้าไม่ได้