Yayımla–Abone Ol

Önceki ders çakışma denetimini tek anahtar üzerinde çalıştırdı: bir istemcinin okuduğu değerin başkası tarafından değiştirilip değiştirilmediği izlemeyle görüldü ve çakışan işlem yeniden denendi. Orada iletişim, aynı anahtara dokunan iki istemci arasında dolaylı biçimde kuruldu. Bir istemcinin başka bir istemciye doğrudan bir şey söylemesi, yani deponun bir iletim kanalı olarak kullanılması hiç ele alınmadı.

Bellek içi depolar bunun için ayrı bir yol taşır: bir istemci bir kanala yayımlar, o kanala abone olan istemciler iletiyi alır. Anahtar alanına hiçbir şey yazılmaz. Bu dersin ölçtüğü şey o yolun hızı değil, neyi güvence altına almadığıdır.

Deponun Yayım Yolu

Kütüphane sisteminde ödünç olayları şube panolarına anlık olarak düşer. Aşağıdaki düzenek üç şube kanalı, her şube için bir pano ve üç yüzüncü adımda katılan bir denetim abonesi kurar. Panoların bağlantısı belirlenimli bir takvimle kopar; gerçek ağ kesintisi yerine adım sayısına bakan bir kural kullanılır, böylece sayılar her koşumda aynı çıkar (BK10).

// yayim-kaybi.mjs — abonesi olmayan kanala yayimlanan ileti kaybolur; kayip sayilir
class Depo {                                       // yayimla-abone ol: kayit yok, geri alma yok
  constructor() { this.kanallar = new Map(); }     // kanal -> Set(abone)
  abone(kanal, a) {
    if (!this.kanallar.has(kanal)) this.kanallar.set(kanal, new Set());
    this.kanallar.get(kanal).add(a);
  }
  birak(kanal, a) { this.kanallar.get(kanal)?.delete(a); }
  yayimla(kanal, ileti) {                          // donen sayi: iletinin ulastigi abone sayisi
    const kume = this.kanallar.get(kanal);
    if (!kume || kume.size === 0) return 0;         // abonesizse ileti hicbir yere yazilmaz
    for (const a of kume) a.al(ileti);
    return kume.size;
  }
}

const olaylar = (adet) => {                        // belirlenimli is yuku, tohum gorunur
  let tohum = 20250731;
  const rast = () => ((tohum = (tohum * 1103515245 + 12345) % 2147483648) / 2147483648);
  return Array.from({ length: adet }, (_, i) => ({
    adim: i + 1, sube: Math.floor(rast() * 3) + 1, kitapId: Math.floor(rast() * 400) + 1,
  }));
};

const depo = new Depo();
const kanal = (s) => `sube:${s}:odunc`;
const panolar = [1, 2, 3].map((s) => ({ ad: `pano${s}`, sube: s, aldi: 0, al() { this.aldi += 1; } }));
const denetim = { ad: "denetim", aldi: 0, al() { this.aldi += 1; } };
const sayac = [1, 2, 3].map(() => ({ yayim: 0, teslim: 0, abonesiz: 0 }));

// Pano baglantisi: her pano donemsel olarak kopar (belirlenimli takvim, rastgelelik yok).
const panoAcik = (s, adim) => (adim + s * 37) % 100 >= 25;   // adimlarin %75'inde bagli
for (const o of olaylar(600)) {
  for (const p of panolar) {                       // baglanti durumu abonelige cevrilir
    const acik = panoAcik(p.sube, o.adim);
    if (acik) depo.abone(kanal(p.sube), p); else depo.birak(kanal(p.sube), p);
  }
  if (o.adim === 301) for (const s of [1, 2, 3]) depo.abone(kanal(s), denetim);  // gec katilan abone

  const s = sayac[o.sube - 1];
  const ulasan = depo.yayimla(kanal(o.sube), o);
  s.yayim += 1; s.teslim += ulasan;
  if (ulasan === 0) s.abonesiz += 1;
}

console.log(["kanal", "yayim", "teslim", "abonesiz yayim", "panonun kacirdigi"]
  .map((b, i) => (i === 0 ? b.padEnd(16) : b.padStart(18))).join(""));
let toplam = [0, 0, 0, 0];
for (const [i, s] of sayac.entries()) {
  const kacan = s.yayim - panolar[i].aldi;
  toplam = toplam.map((t, j) => t + [s.yayim, s.teslim, s.abonesiz, kacan][j]);
  console.log(kanal(i + 1).padEnd(16) +
    [s.yayim, s.teslim, s.abonesiz, kacan].map((n) => String(n).padStart(18)).join(""));
}
console.log("toplam".padEnd(16) + toplam.map((n) => String(n).padStart(18)).join(""));
console.log(`gec katilan abonenin aldigi: ${denetim.aldi}  (katilmadan once yayimlanan: 300)`);

kanal                        yayim            teslim    abonesiz yayim panonun kacirdigi
sube:1:odunc                   199               266                21                46
sube:2:odunc                   195               227                31                55
sube:3:odunc                   206               260                22                46
toplam                         600               753                74               147
gec katilan abonenin aldigi: 300  (katilmadan once yayimlanan: 300)

Üç sayı okunmalı. Teslim sütunu yayım sayısından büyüktür (753 > 600), çünkü bir ileti kaç abone varsa o kadar kopyalanır; yayımlayan taraf bu sayıyı önceden bilmez. Abonesiz yayım sütunu 74 iletinin hiçbir yere yazılmadığını söyler: kanalda o an kimse yoktu, ileti üretildiği adımda yok oldu. Panonun kaçırdığı sütunu daha büyüktür (147), çünkü bir pano bağlı değilken başka bir abone bağlıysa ileti kaybolmaz ama o panoya ulaşmaz.

Son satır en keskin olanıdır. Denetim abonesi üç yüzüncü adımda katıldı ve tam 300 ileti aldı; kendisinden önceki 300 iletiyi hiçbir yolla göremez. Deponun anahtar alanında bu iletilerden geriye hiçbir iz kalmamıştır. Yayımla–abone ol yolunda teslim, iletinin yayımlandığı an ile abonenin bağlı olduğu anın çakışmasına indirgenmiştir.

Yavaş Abonenin Tamponu

Kayıp yalnız bağlantı kopukluğundan gelmez. Abone bağlı olduğu hâlde iletileri okuduğundan hızlı üretiliyorsa, okunmamış iletiler deponun tuttuğu bir çıkış tamponunda birikir. Bu tampon bellekten yenir ve yayımlayanın hesabında görünmez. Aşağıdaki koşum, aboneyi farklı okuma hızlarıyla çalıştırır ve tamponun tuttuğu baytı sayar. Tampon 32.768 baytı aştığında depo aboneyi düşürür; düşürülen abone için sonraki her yayım kayıptır.

// yavas-abone.mjs — yavas abonenin cikis tamponu: tutulan bayt ve sinira carpma adimi
const ILETI = (adim, sube, kitapId) =>            // aynen tasinacak govde; bayti olculur
  JSON.stringify({ olay: "odunc", adim, sube, kitapId });

function kosu(okumaAraligi, sinirBayt, adet = 2000) {
  let tohum = 20250731;
  const rast = () => ((tohum = (tohum * 1103515245 + 12345) % 2147483648) / 2147483648);
  const tampon = [];                               // aboneye yazilamamis iletiler
  let bayt = 0, zirve = 0, kopmaAdimi = 0, kopmaSonrasi = 0, okunan = 0;
  for (let adim = 1; adim <= adet; adim += 1) {
    const g = ILETI(adim, Math.floor(rast() * 3) + 1, Math.floor(rast() * 400) + 1);
    if (kopmaAdimi) { kopmaSonrasi += 1; continue; }   // kopan abone icin yayim kayiptir
    tampon.push(g); bayt += Buffer.byteLength(g);  // depo once tampona yazar
    zirve = Math.max(zirve, bayt);
    if (bayt > sinirBayt) { kopmaAdimi = adim; continue; }   // sinir asilinca abone dusurulur
    if (adim % okumaAraligi === 0) {               // abone her `okumaAraligi` adimda bir okur
      const c = tampon.shift();
      if (c !== undefined) { bayt -= Buffer.byteLength(c); okunan += 1; }
    }
  }
  return { okumaAraligi, okunan, zirve, bekleyen: tampon.length, kopmaAdimi, kopmaSonrasi };
}

console.log(`ileti govdesi: ${ILETI(1, 1, 1)}  (${Buffer.byteLength(ILETI(1, 1, 1))} bayt)`);
console.log(["okuma araligi", "okunan", "zirve bayt", "tamponda kalan", "kopma adimi", "kopma sonrasi"]
  .map((b, i) => (i === 0 ? b.padEnd(15) : b.padStart(15))).join(""));
for (const aralik of [1, 2, 4, 10]) {
  const r = kosu(aralik, 32768);
  console.log(`her ${String(aralik).padStart(2)} adimda bir`.padEnd(15) +
    [r.okunan, r.zirve, r.bekleyen, r.kopmaAdimi, r.kopmaSonrasi]
      .map((n) => String(n).padStart(15)).join(""));
}

ileti govdesi: {"olay":"odunc","adim":1,"sube":1,"kitapId":1}  (46 bayt)
okuma araligi           okunan     zirve bayt tamponda kalan    kopma adimi  kopma sonrasi
her  1 adimda bir           2000             51              0              0              0
her  2 adimda bir            651          32784            653           1304            696
her  4 adimda bir            219          32780            659            878           1122
her 10 adimda bir             73          32804            660            733           1267

Yetişen abone belleğe 51 bayt bindirir: tampon hiçbir zaman bir iletiden fazlasını tutmaz. Okuma aralığı ikiye çıktığında aynı abone 32.784 bayta ulaşır ve 1304. adımda düşürülür; kalan 696 yayım onun için yoktur. Aralık ona çıktığında sınıra 733. adımda çarpılır ve kayıp 1267 iletiye yükselir.

Buradaki bütçe ilişkisi tersine işler. Bir abonenin yavaşlığı deponun belleğinden ödenir ve bu bedel abone başınadır: yüz yavaş abone aynı sınırla yüz ayrı tampon demektir, yani en kötü durumda 3,2 MB. Sınır, deponun kendini koruma biçimidir; bedeli, sınıra çarpan abonenin sessizce düşürülmesidir. Yayımlayan taraf bunun hiçbirini görmez, çünkü yayım çağrısı iletinin okunmasını beklemez (BK11).

Kalıcı Olan ile Olmayan

Aynı olay akışı, deponun akış yapısıyla da kurulabilir. Fark tek bir noktadadır: akış her girişi bir kimlikle saklar, dolayısıyla geç gelen ya da kopup dönen abone kaldığı yerden okuyabilir. Bunun karşılığı bellektir. Aşağıdaki koşum 2000 olayı üç yolla geçirir; abone 800. adımda kopar, 1500. adımda döner. Giriş başına 16 bayt işaretçi/uzunluk payı varsayılır (BK12).

// akis-karsilastirma.mjs — ayni olay akisi: kalici olmayan yayim ile kirpilmis akisin bayti
const EK_YUK = 16;                                 // giris basina isaretci/uzunluk payi (kendi hesabim)
const ILETI = (adim, sube, kitapId) => JSON.stringify({ olay: "odunc", adim, sube, kitapId });
const KOPMA = 800, DONUS = 1500, ADET = 2000;      // abone 800'de kopar, 1500'de geri gelir

function olaylar(adet) {
  let tohum = 20250731;
  const rast = () => ((tohum = (tohum * 1103515245 + 12345) % 2147483648) / 2147483648);
  return Array.from({ length: adet }, (_, i) =>
    ILETI(i + 1, Math.floor(rast() * 3) + 1, Math.floor(rast() * 400) + 1));
}

function yayimYolu() {                             // kalici olmayan: kanal kaydindan baskasi yok
  let bayt = 0;
  for (const ad of ["sube:etkinlik", "pano"]) bayt += Buffer.byteLength(ad) + EK_YUK;
  return { ad: "yayimla-abone ol", donusBayt: bayt, sonBayt: bayt, giris: 0, geriAlinan: 0 };
}

function akisYolu(govdeler, kirpma) {              // kalici: giris kimligi ve govde bellekte durur
  const akis = [];
  let bayt = 0, donusBayt = 0, geriAlinan = 0;
  const olc = (g) => Buffer.byteLength(g.kimlik) + Buffer.byteLength(g.govde) + EK_YUK;
  govdeler.forEach((govde, i) => {
    const giris = { kimlik: `${i + 1}-0`, govde };
    akis.push(giris); bayt += olc(giris);
    while (kirpma && akis.length > kirpma) bayt -= olc(akis.shift());   // en eski giris duser
    if (i + 1 === DONUS) {                         // abone tam bu adimda geri geliyor
      donusBayt = bayt;
      geriAlinan = akis.filter((g) => {            // yalniz bosluktaki girisler geri alinabilir
        const adim = Number(g.kimlik.split("-")[0]);
        return adim >= KOPMA && adim < DONUS;
      }).length;
    }
  });
  return { ad: kirpma ? `akis (kirpma=${kirpma})` : "akis (kirpmasiz)",
           donusBayt, sonBayt: bayt, giris: akis.length, geriAlinan };
}

const govdeler = olaylar(ADET);
const bosluk = DONUS - KOPMA;                      // abonenin bagli olmadigi olay sayisi
console.log(`olay=${ADET}  bosluk=${bosluk} olay (adim ${KOPMA}-${DONUS - 1})`);
console.log(["yol", "bayt@1500", "bayt@2000", "giris", "geri alinan", "kayip"]
  .map((b, i) => (i === 0 ? b.padEnd(20) : b.padStart(13))).join(""));
for (const r of [yayimYolu(), akisYolu(govdeler, 500), akisYolu(govdeler, 0)]) {
  console.log(r.ad.padEnd(20) +
    [r.donusBayt, r.sonBayt, r.giris, r.geriAlinan, bosluk - r.geriAlinan]
      .map((n) => String(n).padStart(13)).join(""));
}

olay=2000  bosluk=700 olay (adim 800-1499)
yol                     bayt@1500    bayt@2000        giris  geri alinan        kayip
yayimla-abone ol               49           49            0            0          700
akis (kirpma=500)           36365        36362          500          499          201
akis (kirpmasiz)           106881       143243         2000          700            0

Üç satır bu kursun kuralını olduğu gibi gösteriyor. Yayımla–abone ol yolu 49 bayt tutuyor; bu, kanal adı ile abone kaydından ibarettir ve olay sayısıyla hiç büyümez. Karşılığında kopan abonenin 700 olayının tamamı kayıptır.

Kırpılmış akış 36.365 bayt tutuyor ve boşluğun 499’unu geri veriyor; kalan 201 olay, abonenin dönüşünden önce kırpıldığı için kaybolmuştur. Kırpma uzunluğu burada doğrudan bir kurtarma penceresidir: 500 giriş, geriye dönük 500 olay demektir.

Kırpmasız akış hiçbir şey kaybetmiyor ama 1500. adımda 106.881, 2000. adımda 143.243 bayt tutuyor. Sayının iki adım arasında büyümesi asıl uyarıdır: kırpması olmayan bir akışın belleği olay sayısıyla sınırsız büyür ve bir bellek sınırı devreye girdiğinde bu büyüme başka verilerin tahliyesine yol açar.

Yayımın Yeri

Bu ölçümler yayımla–abone ol yolunu değersiz kılmaz; yerini belirler. Kaybı kabul edilebilir olan iş, iletinin kaçırılmasının düzeltilebilir olduğu iştir: bir panonun anlık sayacı, bir önbellek geçersizleştirme uyarısı, bir yapılandırma değişikliği duyurusu. Bunların hepsinde abone kaçırdığını bir sonraki tam okumada telafi eder.

Kaybın kabul edilemez olduğu işlerde ölçülen bedel ödenir: ödünç kaydının muhasebeye geçmesi, gecikme cezasının işlenmesi, bir kitabın rafa dönüşünün kaydı. Bu işler ya kalıcı bir akışa yazılır ya da deponun dışına, teslim güvencesi veren bir yapıya bırakılır. Ayrım “hangisi hızlı” sorusuyla değil, “kaçırılan ileti geri getirilebilir mi” sorusuyla yapılır.

Özet

• Yayımla–abone ol yolu iletiyi anahtar alanına yazmaz: 600 yayımın 74’ü o an kanalda abone bulunmadığı için hiçbir yere ulaşmadan yok oldu, panolar ayrıca 147 ileti kaçırdı.
• Teslim sayısı yayım sayısından bağımsızdır; 600 yayım 753 teslim kopyası üretti ve yayımlayan taraf bu sayıyı önceden bilmez.
• Geç katılan abone katılmadan önceki 300 iletinin hiçbirini göremez; geçmiş, deponun hiçbir yerinde durmaz.
• Yavaş abonenin bedeli deponun belleğidir: okuma aralığı ikiye çıktığında çıkış tamponu 32.784 bayta ulaştı, abone 1304. adımda düşürüldü ve sonraki 696 yayım onun için kayboldu.
• Aynı akış kalıcı yapıyla kurulduğunda 49 bayt yerine 36.365 bayt (kırpma 500) ya da 143.243 bayt (kırpmasız) tutulur; satın alınan şey sırasıyla 499 ve 700 olayın geri alınabilmesidir.

Sonraki Adım

Buraya kadarki dersler deponun tek tek yeteneklerini ölçtü: çoğaltma, devralma, bölümleme, çakışma denetimi ve şimdi yayım. Her biri kendi maliyetiyle tanımlandı, ama hiçbiri “bu iş bu depoda hangi yapıyla kurulur ve kaç bayt tutar” sorusuna uçtan uca yanıt vermedi. Sonraki ders kütüphane sisteminin dört tipik işini — katalog önbelleği, oturum kaydı, hız sınırı ve popüler kitap sıralaması — tek tek ele alır; her birini bir yapıya oturtur ve her birinin bellek hesabını aynı tabloda toplar.