Process Substitution ve Named Pipe (mkfifo) ile Dosyalar Arası Gerçek Zamanlı Veri Akışı Yönetimi

Yazar:komut.io ekibi Dosya ve Dizin Yönetimi Komutları
0 Yorum

Gerçek zamanlı veri akışı yönetimi, birçok sysadmin’in yüzeysel geçtiği ama derinlemesine öğrenildiğinde iş hayatını kökten değiştiren bir alan. Ben de yıllarca cat file | grep pattern | awk '{print $1}' zincirleriyle idare ettim, ta ki process substitution ve named pipe’ların gerçek gücünü keşfedene kadar. Bu yazıda size hem teorik altyapıyı hem de “ah, bu işe yararmış!” dedirten pratik senaryoları aktaracağım.

Temel Kavramlar: Anonymous Pipe vs Named Pipe

Bash’te alışkın olduğunuz | (pipe) operatörü aslında anonymous pipe olarak adlandırılır. İki process arasında tek yönlü, geçici bir kanal açar ve işi bitince kaybolur. Bunu biliyorsunuz zaten.

Named pipe (FIFO – First In, First Out) ise dosya sistemi üzerinde gerçek bir varlık olarak duran, isimli bir iletişim kanalıdır. ls -la yaptığınızda p ile başlayan dosya türü olarak görünür. Silinene kadar orada durur ve birden fazla process tarafından kullanılabilir.

Process substitution ise Bash’in size sunduğu bir sözdizimi sihiri: bir komutun çıktısını sanki bir dosyaymış gibi başka bir komuta geçirmenizi sağlar. <(komut) veya >(komut) formunda kullanılır.

Bu üç kavramın birbirinden farkını net anlamadan devam etmek, araçları yanlış yerde kullanmanıza yol açar.

mkfifo ile Named Pipe Oluşturmak

# Basit bir named pipe oluşturma
mkfifo /tmp/veri_kanali

# İzin vererek oluşturma
mkfifo -m 660 /tmp/guvenli_kanal

# Oluşan dosyaya bakalım
ls -la /tmp/veri_kanali
# prw-rw---- 1 admin admin 0 Oca 15 14:23 /tmp/veri_kanali

Buradaki p harfi pipe anlamına gelir. Boyutun 0 olması normal, çünkü FIFO kernel buffer’ında tutar veriyi, dosyada değil.

Şimdi iki terminal açıp şunu deneyin:

# Terminal 1 - okuyucu (önce bunu çalıştırın)
cat /tmp/veri_kanali

# Terminal 2 - yazıcı
echo "Merhaba, named pipe!" > /tmp/veri_kanali

Terminal 1’de mesajı gördünüz mü? İşte bu kadar. Ama fark ettiniz mi: okuyucu process, yazıcı bağlanana kadar bloklandı. Bu, FIFO’nun temel davranışıdır ve bu özelliği hem güç hem de tuzak olabilir.

Process Substitution: <() ve >() Sözdizimi

Process substitution, Bash 3.x ve üzerinde çalışır (eski sh veya dash’te çalışmaz, dikkat edin).

# İki dosyayı sıralayıp diff almak - geleneksel yol
sort dosya1.txt > /tmp/sorted1
sort dosya2.txt > /tmp/sorted2
diff /tmp/sorted1 /tmp/sorted2
rm /tmp/sorted1 /tmp/sorted2

# Process substitution ile tek satırda
diff <(sort dosya1.txt) <(sort dosya2.txt)

Perde arkasında ne oluyor? Bash /dev/fd/63 veya /proc/self/fd/63 gibi bir dosya tanımlayıcısı oluşturuyor ve sort komutunun çıktısını oraya yönlendiriyor. diff bu dosya tanımlayıcısını normal bir dosya gibi okuyor.

# /dev/fd üzerinden ne göründüğünü anlamak için
ls -la <(echo test)
# lr-x------ 1 user user 64 Oca 15 14:25 /dev/fd/63 -> pipe:[12345]

Gördüğünüz gibi, aslında bir pipe’a işaret eden sembolik link. Kernel bunu otomatik yönetiyor.

>() ile Çıktı Yönlendirme

# Bir log dosyasını hem ekrana hem dosyaya yönlendirmek
# tee'nin process substitution ile kullanımı
make 2>&1 | tee >(grep -i error > errors.log) >(grep -i warning > warnings.log)

# Aynı veriyi farklı formatlarda kaydetmek
backup_script | tee >(gzip > /backup/full.tar.gz) >(sha256sum > /backup/full.sha256)

Bu son örnek gerçekten güçlü: backup işleminin çıktısını hem sıkıştırılmış dosyaya hem de SHA256 hesaplamasına aynı anda gönderiyoruz. Veriyi iki kez işlemek zorunda kalmıyoruz.

Gerçek Dünya Senaryosu 1: Log Analiz Pipeline’ı

Bir üretim ortamında şu problemi yaşadım: Nginx access log’larını gerçek zamanlı analiz etmem gerekiyordu ama aynı anda hem anomali tespiti hem de istatistik toplama yapmak istiyordum.

# /tmp/log_pipe adında bir FIFO oluştur
mkfifo /tmp/log_pipe

# Terminal 1: Log akışını FIFO'ya yönlendir
tail -f /var/log/nginx/access.log > /tmp/log_pipe &

# Terminal 2: FIFO'dan oku ve işle
while IFS= read -r line; do
    # 5xx hataları için alarm
    if echo "$line" | grep -q '" 5[0-9][0-9] '; then
        echo "[ALARM] $(date): $line" >> /var/log/alarm.log
        # Slack bildirimi veya mail de gönderilebilir
    fi
    
    # İstatistik için
    echo "$line" | awk '{print $9}' >> /tmp/status_codes.tmp
done < /tmp/log_pipe

Ancak bu yaklaşımın bir sorunu var: tek okuyucu. Birden fazla process aynı FIFO’dan okumaya çalışırsa veri bölünür. Bunun çözümü için farklı bir mimari gerekir.

Gerçek Dünya Senaryosu 2: Parallel Veri İşleme

Process substitution’ın öne çıktığı yer: aynı veri akışını birden fazla işleme paralel olarak göndermek.

#!/bin/bash
# Büyük CSV dosyasını hem valide et hem işle hem de arşivle

process_large_csv() {
    local input_file="$1"
    
    cat "$input_file" | tee 
        >(awk -F',' 'NR>1 && NF!=5 {print "Hatalı satır:", NR, $0}' > validation_errors.log) 
        >(tail -n +2 | awk -F',' '{sum+=$3} END {print "Toplam:", sum}' > totals.txt) 
        >(gzip > "archived_$(date +%Y%m%d).csv.gz") 
        > /dev/null
    
    echo "İşlem tamamlandı"
    echo "Hata sayısı: $(wc -l < validation_errors.log)"
    cat totals.txt
}

process_large_csv data/satis_raporu.csv

Bu script’te veri tek kez okunuyor ama üç farklı yere aynı anda akıyor. Disk I/O ve işlem süresi açısından büyük kazanım.

Named Pipe ile Producer-Consumer Pattern

Sistem yönetiminde sıkça karşılaşılan bir senaryo: bir kaynak veri üretiyor, başka bir process bu veriyi tüketiyor. Named pipe bu pattern için biçilmiş kaftan.

#!/bin/bash
# Basit bir job queue implementasyonu

QUEUE_PIPE="/tmp/job_queue"
RESULT_PIPE="/tmp/job_results"

# Cleanup fonksiyonu
cleanup() {
    rm -f "$QUEUE_PIPE" "$RESULT_PIPE"
    echo "Temizlendi"
}
trap cleanup EXIT

# FIFO'ları oluştur
mkfifo "$QUEUE_PIPE" "$RESULT_PIPE"

# Worker process - arka planda
worker() {
    while IFS= read -r job; do
        [[ "$job" == "STOP" ]] && break
        
        echo "İşleniyor: $job"
        # Simüle edilmiş iş
        result=$(echo "$job" | tr '[:lower:]' '[:upper:]')
        sleep 0.5
        
        echo "$result" > "$RESULT_PIPE"
    done < "$QUEUE_PIPE"
} 

worker &
WORKER_PID=$!

# Sonuç okuyucu
result_reader() {
    local count=0
    while IFS= read -r result; do
        echo "Sonuç $((++count)): $result"
        [[ $count -ge 3 ]] && break
    done < "$RESULT_PIPE"
}

result_reader &

# İş gönder
for job in "dosya_isle" "backup_al" "rapor_olustur"; do
    echo "$job" > "$QUEUE_PIPE"
    echo "Gönderildi: $job"
done

echo "STOP" > "$QUEUE_PIPE"
wait $WORKER_PID
echo "Worker tamamlandı"

Bloklama Sorunları ve Çözümleri

Named pipe kullanırken en sık karşılaşılan sorun bloklama. Okuyucu olmadan yazmaya çalışmak, ya da yazıcı olmadan okumaya çalışmak process’i askıya alır.

# Sorun: Bloklanan write
echo "veri" > /tmp/bos_pipe  # Okuyucu yoksa burada takılır!

# Çözüm 1: Non-blocking open
# exec ile file descriptor aç
exec 3<>/tmp/log_pipe  # Hem okuma hem yazma için aç (bloklamaz)
echo "test" >&3
cat <&3

# Çözüm 2: Arka plan process ile
(echo "veri" > /tmp/bos_pipe) &
# Okuyucu hazır olduğunda arka plan process devam eder

# Çözüm 3: /dev/null ile birlikte aç
exec 3<> /tmp/log_pipe
echo "veri" >&3 &
# Yazma tarafı için okuyucu simülasyonu

Daha pratik bir çözüm, okuyucu ve yazıcıyı neredeyse eşzamanlı başlatmak:

#!/bin/bash
mkfifo /tmp/data_stream

# Okuyucuyu arka plana al
grep "ERROR" /tmp/data_stream > errors.txt &
READER_PID=$!

# Şimdi yaz (okuyucu hazır)
journalctl -f --since "1 hour ago" > /tmp/data_stream

wait $READER_PID
rm /tmp/data_stream

diff, comm ve join ile Process Substitution Kullanımı

Bu araçlar normalde dosya argümanı bekler, process substitution onları akış tabanlı kullanıma açar.

# İki sunucunun kurulu paket listesini karşılaştır
diff <(ssh sunucu1 "dpkg -l | awk '{print $2}' | sort") 
     <(ssh sunucu2 "dpkg -l | awk '{print $2}' | sort")

# Her iki sunucuda da olan paketler
comm -12 
    <(ssh sunucu1 "dpkg --get-selections | sort") 
    <(ssh sunucu2 "dpkg --get-selections | sort")

# Aktif process listelerini karşılaştır
diff <(ps aux | awk '{print $11}' | sort -u) 
     <(cat /tmp/expected_processes.txt | sort -u)

Bu tür karşılaştırmaları geçmişte geçici dosyalarla yapardım. Şimdi tek satırda hallediyor ve temizlik derdi kalmıyor.

Gerçek Dünya Senaryosu 3: Backup Doğrulama Pipeline’ı

Bu senaryoyu bir müşteri projesinde geliştirdim. Backup alırken aynı anda sıkıştırma, checksum hesaplama ve şifreleme yapmak gerekiyordu.

#!/bin/bash
# Paralel backup pipeline

BACKUP_DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_DIR="/backup"
SOURCE_DIR="/data/production"

backup_with_verification() {
    local source="$1"
    local dest_base="$2"
    
    echo "Backup başlatılıyor: $source"
    
    # tar çıktısını üç farklı yere gönder:
    # 1. Sıkıştırılmış backup dosyası
    # 2. SHA256 checksum
    # 3. Dosya listesi
    tar cf - "$source" | tee 
        >(gzip -9 > "${dest_base}.tar.gz") 
        >(sha256sum | awk '{print $1}' > "${dest_base}.sha256") 
        >(tar tf - > "${dest_base}.filelist") 
        > /dev/null
    
    # Boyut kontrolü
    local backup_size=$(stat -c%s "${dest_base}.tar.gz" 2>/dev/null || echo 0)
    local source_size=$(du -sb "$source" 2>/dev/null | awk '{print $1}')
    
    echo "Kaynak boyutu: $(numfmt --to=iec $source_size)"
    echo "Backup boyutu: $(numfmt --to=iec $backup_size)"
    echo "Sıkıştırma oranı: $(echo "scale=1; $backup_size * 100 / $source_size" | bc)%"
    echo "Checksum: $(cat "${dest_base}.sha256")"
    echo "Dosya sayısı: $(wc -l < "${dest_base}.filelist")"
}

backup_with_verification "$SOURCE_DIR" "$BACKUP_DIR/prod_$BACKUP_DATE"

Process Substitution ile Hata Yönetimi

Sık atlanan bir konu: process substitution içindeki komutların exit kodları nasıl yakalanır?

#!/bin/bash
# Hata yönetimi örneği

# PIPESTATUS ile pipe hataları
validate_and_process() {
    local input="$1"
    
    # Her aşamada hata kontrolü
    cat "$input" | 
        grep -v "^#" | 
        awk '{print $1, $2}' | 
        sort -k1,1 > /tmp/processed.txt
    
    # Pipe'daki her komutun exit kodunu kontrol et
    for status in "${PIPESTATUS[@]}"; do
        if [[ $status -ne 0 ]]; then
            echo "Pipeline başarısız, exit kod: $status" >&2
            return 1
        fi
    done
    
    echo "Başarıyla işlendi"
}

# Process substitution hatalarını yakalamak daha zor
# Geçici değişken kullanımı önerilir
temp_result=$(mktemp)
diff <(sort "$1" 2>"$temp_result") <(sort "$2" 2>>"$temp_result")
diff_exit=$?

if [[ -s "$temp_result" ]]; then
    echo "Hatalar:" >&2
    cat "$temp_result" >&2
fi
rm -f "$temp_result"
exit $diff_exit

Performans: Ne Zaman Hangisini Kullanmalı

Pratikte öğrendiğim bazı kural niteliğindeki bilgiler:

  • Process substitution kullanın: Tek seferlik işlemler için, birden fazla komutu dosya argümanı bekleyen bir araca geçirmek için, geçici dosya yaratmaktan kaçınmak istediğinizde.
  • Named pipe (mkfifo) kullanın: Farklı process’ler arasında uzun süreli iletişim kurmanız gerektiğinde, farklı zamanlarda başlayan ve biten producer-consumer senaryolarında, script’ler arası koordinasyon için.
  • Anonymous pipe (|) kullanın: Basit zincirleme için, aynı script içindeki sıralı işlemler için.
# Hız testi - büyük veri seti için
time (
    # Geçici dosya yöntemi
    sort /dev/urandom | head -1000000 > /tmp/sorted1.tmp
    sort /dev/urandom | head -1000000 > /tmp/sorted2.tmp
    diff /tmp/sorted1.tmp /tmp/sorted2.tmp > /dev/null
    rm /tmp/sorted1.tmp /tmp/sorted2.tmp
)

time (
    # Process substitution yöntemi
    diff <(sort /dev/urandom | head -1000000) 
         <(sort /dev/urandom | head -1000000) > /dev/null
)

Çoğu durumda process substitution hem daha hızlı hem de daha az kaynak tüketir, çünkü disk yazımı gereksiz hale gelir.

Dikkat Edilmesi Gereken Durumlar

Birkaç tuzağı paylaşayım, bunları zor yoldan öğrendim:

# YANLIŞ: sh ile çalışmaz!
sh -c 'diff <(ls /tmp) <(ls /var)'  # syntax error

# DOĞRU: bash kullan
bash -c 'diff <(ls /tmp) <(ls /var)'

# YANLIŞ: Named pipe silinmeden script çıkarsa kalır
mkfifo /tmp/kanal
# ... script patlarsa /tmp/kanal kalır

# DOĞRU: trap ile temizle
PIPE=/tmp/kanal_$$
mkfifo "$PIPE"
trap "rm -f $PIPE" EXIT INT TERM

# YANLIŞ: Okuyucusuz named pipe'a yazma girişimi
echo "veri" > /tmp/kanal  # sonsuza kadar bekler

# DOĞRU: Timeout ile koruma
timeout 5 bash -c 'echo "veri" > /tmp/kanal' || echo "Zaman aşımı"

Sonuç

Process substitution ve named pipe, Unix felsefesinin “her şeyi dosya say” prensibinin en güzel tezahürlerinden. Bu araçları kavradıktan sonra gereksiz geçici dosya yaratmayı bıraktım, pipeline’larım daha okunur hale geldi ve özellikle büyük veri kümeleriyle çalışırken ciddi performans kazanımları elde ettim.

Yeni başlayanlar için tavsiyem: önce mkfifo ile iki terminal arasında veri akışını elle deneyin. Bloklama davranışını hissedin. Sonra process substitution’ı basit diff <(komut1) <(komut2) örnekleriyle oturtun. Oradan üst senaryolara geçmek çok daha kolay olacak.

Bu kavramlar özellikle log analizi, backup pipeline’ları ve monitoring sistemleri kuruyorsanız doğrudan işinize yarayacak. Geçici dosyaların yarattığı disk baskısını, temizlik zahmetini ve race condition risklerini ortadan kaldırıyor. Sisteminizin dosya sistemi değil, veri akışları üzerinden çalıştığını düşünmeye başladığınızda bu araçların değeri çok daha net görünüyor.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir