Node.js stream.pipeline ve Backpressure: Stream'leri Güvenle Bağlamak

Backpressure nedir ve neden önemlidir?

Node.js Learn'deki "Backpressuring in Streams" rehberi sorunu bir tıkanma benzetmesiyle anlatır: veri aktarımında alıcı taraf yavaşsa, kaynaktan gelen veri tamponun arkasında birikir. Diskten okuma diske yazmaktan hızlı olduğu için, bir dosyayı sıkıştırıp başka bir dosyaya yazarken bu durum kaçınılmazdır. Bir akış kontrol mekanizması olmasaydı biriken veri belleği doldurur, çöp toplayıcıyı (garbage collector) yorar ve diğer süreçleri yavaşlatırdı.

Rehber bunu bir deneyle gösterir: büyük bir dosyayı gzip ile sıkıştıran aynı betik, write() dönüş değerine uyan normal Node ile yaklaşık 87,81 MB, bu kontrol devre dışı bırakılmış değiştirilmiş bir Node ile yaklaşık 1,52 GB en yüksek bellek kullanımına ulaşmıştır. Rehberdeki ölçümlerde çöp toplayıcının davranışı da backpressure olmayan sürümde belirgin biçimde kötüleşir. Bu sayılar rehberin kendi deneyine aittir; sizin ortamınızda farklı olabilir, ama farkın büyüklük mertebesi mekanizmanın neden önemli olduğunu gösterir.

write() false döndüğünde: highWaterMark ve 'drain'

Backpressure'ın tetiklendiği an, belgelere göre tam olarak Writable stream'in write() dönüş değeridir. İç tampon highWaterMark eşiğine ulaşır ya da aşarsa write() false döner. Bu bir hata değildir; veri kabul edilmiş ve kuyruğa alınmıştır, ama stream sizden 'drain' olayı gelene kadar yazmayı bırakmanızı istemektedir. Aşağıdaki örnekte bilerek 10 baytlık bir eşik ve her parçayı 10 ms'de işleyen yavaş bir hedef kurduk:

import { Writable } from 'node:stream';

const yavas = new Writable({
  highWaterMark: 10, // 10 bayt
  write(parca, _enc, callback) {
    setTimeout(callback, 10); // yavaş bir hedef
  },
});

console.log(yavas.write('12345')); // çıktı: true
console.log(yavas.write('67890')); // çıktı: false
console.log(yavas.writableLength, yavas.writableNeedDrain); // çıktı: 10 true

yavas.once('drain', () => {
  console.log('drain geldi, kuyruk:', yavas.writableLength); // çıktı: drain geldi, kuyruk: 0
});

İlk 5 bayt eşiğin altında kaldığı için true, ikinci 5 bayt kuyruğu 10 bayta, yani eşiğe çıkardığı için false döndü. writableNeedDrain özelliği aynı bilgiyi sonradan okumanızı sağlar. Belgeler burada önemli bir uyarı yapar: false döndükten sonra yazmaya devam etmek yasak değildir, Node gelen her parçayı tamponlamayı sürdürür. Ancak bu, bellek tükenene kadar devam edebilir; yüksek bellek kullanımı çöp toplayıcının performansını bozar ve RSS'i büyütür. Uzak tarafın veriyi okumadığı bir TCP soketi hiç boşalmayabileceği için, boşalmayan bir sokete yazmak belgelerin ifadesiyle uzaktan istismar edilebilir bir açığa dönüşebilir.

Elle yazarken doğru desen: once(stream, 'drain')

Veriyi bir döngüde kendiniz üretip yazıyorsanız backpressure'a siz uymalısınız. En okunaklı yol, write() false döndüğünde node:events modülünün once() fonksiyonuyla 'drain' olayını beklemektir. once() bir Promise döndürür, böylece döngü await ile durur. Promise ve async/await kavramları için async/await yazısına, events.once ayrıntısı için EventEmitter yazısına bakabilirsiniz.

import { createWriteStream } from 'node:fs';
import { once } from 'node:events';

const hedef = createWriteStream('sayilar.txt', { highWaterMark: 1024 });
let bekleme = 0;

for (let i = 0; i < 1000; i++) {
  if (!hedef.write(`satır ${i}\n`)) {
    bekleme++;
    await once(hedef, 'drain');
  }
}
hedef.end();
await once(hedef, 'finish');
console.log('drain beklendi:', bekleme > 0); // çıktı: drain beklendi: true

Kaç kez beklendiği diskin hızına bağlı olduğu için kesin sayıyı yazdırmadık; önemli olan, döngünün tampon dolduğunda gerçekten durmasıdır. Belgeler, veri isteğe bağlı üretilebiliyorsa bu mantığı bir Readable stream içine koyup pipe() ya da pipeline() ile bağlamanın daha iyi olduğunu da söyler. Bunun en kolay yolu, üreticiyi bir async generator olarak yazıp pipeline()'a vermektir; aşağıda göreceğiz.

Kendi Readable stream'inizi yazıyorsanız kural öbür taraftan işler: Learn rehberindeki altın kurallara göre istenmeden push() yapmamalı, push() false döndürdüğünde de kaynaktan okumayı bırakmalısınız. Ayrıca readable.on('data', (d) => writable.write(d)) biçimindeki kısa yol, hedefin hazır olup olmadığına bakmadan veri bastığı için rehberde kötü örnek olarak gösterilir.

pipe() ile pipeline() arasındaki fark

readable.pipe(writable) backpressure'ı otomatik yönetir; Learn rehberinin dediği gibi çoğu geliştirici bu yüzden backpressure'ı hiç duymaz. Ancak pipe()'ın belgelerde açıkça yazan bir zayıflığı vardır: kaynak bir hata yayarsa hedef otomatik olarak kapatılmaz. Bellek sızıntısını önlemek için her stream'i elle kapatmanız gerekir. Aşağıdaki örnekte olmayan bir dosyayı okumaya çalışıyoruz:

import { createReadStream, createWriteStream } from 'node:fs';

const kaynak = createReadStream('olmayan-dosya.txt');
const hedef = createWriteStream('kopya2.txt');

kaynak.on('error', (err) => {
  console.log('kaynak hatası:', err.code); // çıktı: kaynak hatası: ENOENT
  setTimeout(() => {
    console.log('hedef kapatıldı mı:', hedef.destroyed); // çıktı: hedef kapatıldı mı: false
    hedef.destroy();
  }, 50);
});
kaynak.pipe(hedef);

Kaynak hata verdi ama hedef açık kaldı; kapatmayı biz yapmak zorunda kaldık. Aynı işi pipeline() ile yaptığınızda hata Promise'in reddi olarak elinize gelir ve hedef stream de yok edilir:

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';

const hedef = createWriteStream('kopya.txt');
try {
  await pipeline(createReadStream('olmayan-dosya.txt'), hedef);
} catch (err) {
  console.log(err.code, err.syscall); // çıktı: ENOENT open
}
console.log('hedef kapatıldı mı:', hedef.destroyed); // çıktı: hedef kapatıldı mı: true

Belgelere göre pipeline(), hata durumunda zincirdeki tüm stream'lerde destroy(err) çağırır; yalnızca 'end' veya 'close' yaymış Readable'lar ve 'finish' veya 'close' yaymış Writable'lar bunun dışındadır. Learn rehberi de eski pump paketine Node 10 ve sonrasında gerek kalmadığını, yerini pipeline'ın aldığını belirtir. ENOENT gibi sistem hatası kodlarının anlamı için sistem hataları yazısına bakabilirsiniz.

Async generator ile dönüştürme

pipeline() kaynak, ara adım ve hedef olarak yalnızca stream değil fonksiyon da kabul eder. Ara adım olarak verdiğiniz async generator, önceki adımın çıktısını source parametresiyle async iterable olarak alır ve yield ettiği değerler bir sonraki adıma gider. Learn rehberi bu yolu ayrı bir Transform sınıfı yazmaktan daha okunaklı bulur:

import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { readFile, writeFile } from 'node:fs/promises';

await writeFile('giris.txt', 'merhaba dünya\nnode stream\n');

await pipeline(
  createReadStream('giris.txt', { encoding: 'utf8' }),
  async function* (kaynak) {
    for await (const parca of kaynak) {
      yield parca.toLocaleUpperCase('tr-TR');
    }
  },
  createWriteStream('cikis.txt'),
);

console.log(await readFile('cikis.txt', 'utf8'));
// çıktı: MERHABA DÜNYA
// çıktı: NODE STREAM

Generator bir sonraki parçayı ancak hedef hazır olduğunda ister, bu yüzden backpressure burada da korunur. Türkçe büyük harf dönüşümü için toLocaleUpperCase('tr-TR') kullandık; bu dil düzeyindeki ayrıntı string metotları yazısında anlatılıyor. Belgeler, generator kaynak olarak (zincirin ilk elemanı) kullanıldığında ikinci parametre olarak gelen { signal } nesnesini işlemenizi, aksi halde pipeline'ın iptal edildiğinde hiç tamamlanmayabileceğini hatırlatır.

İptal ve zaman aşımı: options.signal

Promise tabanlı pipeline(), son argüman olarak { signal } içeren bir seçenek nesnesi alır. Sinyal tetiklendiğinde zincirdeki stream'ler AbortError ile yok edilir. AbortSignal.timeout(ms) ile belirli bir süreden uzun süren aktarımları kesebilirsiniz. Aşağıdaki kaynak hiç bitmeyen bir generator; 100 ms sonra sinyal tetikleniyor:

import { pipeline } from 'node:stream/promises';
import { Readable, Writable } from 'node:stream';

async function* sonsuz() {
  let i = 0;
  while (true) {
    yield `parça ${i++}\n`;
    await new Promise((r) => setTimeout(r, 20));
  }
}

const bos = new Writable({ write(_p, _e, cb) { cb(); } });

try {
  await pipeline(Readable.from(sonsuz()), bos, { signal: AbortSignal.timeout(100) });
} catch (err) {
  console.log(err.name, err.code); // çıktı: AbortError ABORT_ERR
}

Seçenek nesnesinde bir de end vardır: false verirseniz kaynak bittiğinde hedef stream kapatılmaz, böylece aynı hedefe sonra başka veri yazabilirsiniz (Transform stream'ler bu ayardan bağımsız olarak her zaman sonlandırılır). AbortController ve sinyallerin genel kullanımı AbortController ve zaman aşımı yazısında ayrıca anlatılıyor.

Kendi stream'inizi yazarken backpressure kuralları

Hazır stream'leri bağlarken backpressure'ı Node yönetir; kendi Writable ya da Transform sınıfınızı yazdığınızda ise iç durum makinesi yine çoğu işi üstlenir ama birkaç kurala uymanız gerekir. Learn rehberinin Writable için verdiği ilk uyarı, _write() içinde callback'in her parça için tam olarak bir kez çağrılmasıdır. Birden fazla if dalında callback çağırıp sonra dalların dışında bir kez daha çağırırsanız, aynı parça için callback iki kez çalışır; rehber her dalda return callback() yazmayı önerir. Asenkron bir işlemin sonucunu bekliyorsanız callback'i ancak o işlem bittiğinde çağırın; erken çağırmak, hedefin aslında hazır olmadığı halde yeni veri istemesine yol açar.

Transform stream'lerde iki ayrı tampon olduğunu da unutmayın: yazılan taraf ve okunan taraf. Rehbere göre her iki tarafın highWaterMark değeri ayarlanabilir ve backpressure davranışını etkiler. Transform'un çıktısını kimse okumuyorsa okunan taraftaki tampon dolar, ardından yazılan taraf da dolar ve write() false döner. Belgeler bu yüzden Transform stream'lerin bir yere bağlanana ya da 'data' veya 'readable' dinleyicisi eklenene kadar duraklatılmış kaldığını, bu durumdayken yazmanın özellikle sorunlu olduğunu belirtir. Son olarak rehber cork() ve uncork() için bir uyarı yapar: art arda birkaç write() çağrısını tek seferde göndermek için cork() kullanılır ve uncork(), cork() kadar çağrılmalıdır; aksi halde akış yeniden başlamaz.

Callback sürümü ve HTTP ile kullanırken dikkat

node:stream içindeki eski pipeline(...streams, callback) sürümü de geçerlidir ve son stream'i döndürür. Belgelere göre v14.0.0 ile callback, uygun stream'lerde 'close' olayını bekledikten sonra çağrılır hale geldi. İki sürümün de bir yan etkisi vardır: hata sonrasında stream'lerde asılı olay dinleyicileri bırakabilirler; stream'leri hatadan sonra yeniden kullanmak dinleyici sızıntısına ve yutulan hatalara yol açabilir.

HTTP sunucusunda yanıtı pipeline(dosya, res) ile gönderirken belgelerdeki uyarıyı aklınızda tutun: pipeline hata olduğunda tüm stream'leri, dolayısıyla yanıtı ve soketini de kapatır. Bu yüzden dosya bulunamadığında hata callback'inde res.end('hata') göndermeye çalışmak işe yaramaz. Doğru yaklaşım, dosyanın varlığını ve durumunu yanıtı başlatmadan önce kontrol etmek ya da hata halinde yalnızca kaydı tutmaktır. Somut bir örneği HTTP sunucusu yazısında bulabilirsiniz.

Uygulamalı mini örnek: logları sıkıştırarak arşivlemek

Günlük bir erişim logunu gzip ile sıkıştırıp arşiv klasörüne yazmak tipik bir pipeline işidir: createReadStream(log), createGzip() ve createWriteStream(log + '.gz') üç adım olarak verilir. Okuma, sıkıştırma ve yazma hızları farklı olsa da backpressure sayesinde bellek kullanımı sabit kalır; herhangi bir adım hata verirse (disk dolu, dosya yok) diğerleri kapatılır ve hata tek bir catch bloğunda yakalanır. Bu örneğin çalıştırılmış hali ve sıkıştırma ayarları zlib yazısında yer alıyor.

Sık yapılan hatalar

• write() dönüş değerini yok saymak: Döngüde sınırsız write() belleği şişirir. false gelince 'drain' bekleyin.
• pipe() zincirinde yalnızca son stream'i dinlemek: Ara stream'lerin hataları ne iletilir ne de diğer stream'ler kapatılır. pipeline() kullanın.
• Promise sürümünü await etmeden bırakmak: Hata yakalanmamış Promise reddine dönüşür; unhandledRejection yazısına bakın.
• Generator kaynakta signal'i işlememek: İptal edilen pipeline tamamlanmayabilir.
• Hata sonrası stream'leri yeniden kullanmak: Asılı dinleyiciler sızıntı ve yutulan hata üretir; yeni stream oluşturun.
• HTTP'de hata callback'inden yanıt göndermeye çalışmak: pipeline soketi çoktan kapatmış olabilir.

Sık Sorulan Sorular

pipeline callback sürümünü mü Promise sürümünü mü kullanmalıyım?

Yeni kodda node:stream/promises içindeki Promise sürümü async/await ile daha okunaklıdır ve signal seçeneğini doğrudan alır. Callback sürümü eski kodla uyum ve döndürdüğü son stream'e erişim gerektiğinde kullanılır.

Varsayılan highWaterMark nedir?

API belgesine göre stream.getDefaultHighWaterMark() nesne modunda 16, bayt stream'lerinde Windows dışında 65536, Windows'ta 16384 döndürür. Learn rehberinde hâlâ 16 KB ifadesi geçer; güncel değer için API belgesini esas alın. fs.createReadStream'in kendi varsayılanı 64 * 1024 bayttır.

Backpressure varsa veri kaybolur mu?

Hayır. Backpressure veriyi atmaz, üreticiyi yavaşlatır. write() false döndürse bile o parça kuyruğa alınmıştır; yalnızca sonraki yazmalar için beklemeniz istenir.

stream.compose ne işe yarar?

Birden fazla stream'i tek bir stream olarak birleştirir; elde edilen nesneyi başka bir pipeline'da tek adım gibi kullanabilirsiniz. YAML kayıtlarına göre stream.compose() v26.2.0 ile kararlı olarak işaretlendi. Bu yazı en yaygın kullanılan pipeline() üzerine odaklanır.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: Node.js öğrenme rehberi · Node.js stream nedir · zlib gzip ve brotli · AbortController ile iptal · node:http ile sunucu · Node.js performans optimizasyonu