Python में पाइप से लाइन-दर-लाइन पढ़ना और EOF संभालना

जानें कैसे Python में पाइप से लाइन-दर-लाइन टेक्स्ट पढ़ें और writer बंद होने पर EOF पहचानें—select जैसी लो-लेवल जाँच के बिना, सिर्फ readline से शटडाउन

पाइप से पंक्तियाँ पढ़ना और यह पहचानना कि writer कब बंद हो गया है, ऊपर से सीधा लगता है, लेकिन इसे जरूरत से ज्यादा जटिल बनाना आसान है। एक आम गलती है लो-लेवल readiness चेक को हाई-लेवल फ़ाइल-सदृश I/O के साथ मिलाना, और साफ़ end-of-stream संकेत को खो देना। अच्छी बात यह है कि फ़ाइल इंटरफ़ेस पहले से ही इसका जवाब देता है—और यह जितना दिखता है उससे सरल है।

समस्या की रूपरेखा

उद्देश्य है पाइप से टेक्स्ट को लाइन-दर-लाइन पढ़ना, जबकि प्रोग्राम का दूसरा हिस्सा उसमें छोटे-छोटे टुकड़े लिखता रहता है। पेचीदा हिस्सा यह है कि लिखने वाला सिरा कब बंद हुआ, इसे पहचानकर पढ़ने वाला लूप साफ़ तरीके से बाहर निकले। नीचे के उदाहरण में पढ़ने का लूप कभी समाप्त नहीं होता और join तब तक रुका रहता है जब तक मैन्युअल रूप से बाधित न किया जाए।

from datetime import datetime
from itertools import batched
import os
from select import select
from threading import Thread
from time import sleep
# प्रत्येक superscript अंक 3 बाइट में एनकोड होता है।
sample_txt = '⁰¹²\n³\n⁴\n⁵⁶\n⁷⁸⁹⁰¹²\n³'
sample_buf = bytes(sample_txt, 'utf8')
r_fd, w_fd = os.pipe()
bin_out = open(w_fd, 'wb', buffering=0)
txt_in = open(r_fd, 'r')
collected = []
def pump_reader():
    t_prev = datetime.now()
    while True:
        sleep(1)
        t_now = datetime.now()
        print('A', (t_now - t_prev).total_seconds())
        t_prev = t_now
        r_ready, w_ready, e_ready = select([txt_in], [txt_in], [txt_in], 0)
        if txt_in.closed:
            break
        if e_ready:
            break
        if not r_ready:
            continue
        piece = txt_in.readline()
        print('B', (t_now - t_prev).total_seconds())
        t_prev = t_now
        if piece:
            print('got chunk', repr(piece))
            collected.append(piece)
worker = Thread(target=pump_reader)
worker.start()
for segment in batched(sample_buf, 4):
    payload = bytes(segment)
    sleep(1.6)
    bin_out.write(payload)
bin_out.close()
worker.join()
print(repr(collected))

देखा गया व्यवहार: जितनी पंक्तियाँ अपेक्षित थीं, सभी मिलती हैं—आख़िरी पंक्ति बिना trailing newline के भी—लेकिन reader लूप कभी खत्म नहीं होता और थ्रेड जीवित रहता है।

असल में हो क्या रहा है

जब पाइप का लिखने वाला सिरा बंद हो जाता है और आगे कोई डेटा नहीं बचता, तो लाइन-उन्मुख पढ़ाई (readline के जरिए) एक खाली स्ट्रिंग लौटाती है। यह खाली स्ट्रिंग टेक्स्ट-मोड फ़ाइल ऑब्जेक्ट्स के लिए फ़ाइल-के-अंत का संकेत है। दूसरे शब्दों में, अंतिम आंशिक पंक्ति मिल जाने के बाद अगली readline कॉल "" देती है। अगर लूप इसे जांचे बिना चलता रहा, तो वह कभी समाप्त नहीं होगा।

साफ़ संकेत फ़ाइल-सदृश लेयर पर पहले से मौजूद है। इस स्थिति में select, closed फ़्लैग या exception सूचियों को मिलाने की जरूरत नहीं पड़ती। कुंजी यह है कि readline से आई खाली स्ट्रिंग को स्ट्रीम के अंत के रूप में स्वीकारें।

समाधान

readline से लौटे मान की truthiness का इस्तेमाल करें। जैसे ही यह खाली स्ट्रिंग लौटाए, लूप स्वाभाविक रूप से खत्म हो जाएगा। इससे पढ़ने वाला लूप छोटा और सही बनता है।

def pump_reader():
    while line := txt_in.readline():
        print('got chunk', repr(line))
        collected.append(line)

यह रूप EOF अनुबंध को स्पष्ट और भरोसेमंद बनाता है। एक बराबरी का तरीका है फ़ाइल ऑब्जेक्ट पर सीधे iteration करना—वह स्ट्रीम खत्म होने तक पंक्तियाँ देता रहता है।

यह क्यों मायने रखता है

जब आप subprocess पाइप्स या किसी भी producer-consumer पाइपलाइन के साथ काम करते हैं, तो स्थिर अवस्था के throughput जितना ही एक सुदृढ़ शटडाउन पथ भी ज़रूरी होता है। फ़ाइल API के end-of-file व्यवहार पर भरोसा करने से कोड सरल रहता है और वे सूक्ष्म अटके हुए हालात नहीं बनते जहाँ थ्रेड्स हमेशा प्रतीक्षा में रहें और joins तब तक रुके रहें जब तक ज़बरदस्ती रोका न जाए। अगर आप पहले से readline से टेक्स्ट पढ़ रहे हैं, तो उसे स्ट्रीम का अंत खुद बताने दें।

निष्कर्ष

फ़ाइल इंटरफ़ेस से टकराव मत करें। अगर रीडर readline का उपयोग करता है, तो खाली स्ट्रिंग को अंत मानें—लूप बिना अतिरिक्त जाँच या टाइमआउट के बाहर आ जाएगा। इससे जटिलता घटती है, आंशिक पंक्तियों के आसपास के किनारे मामलों में कमी आती है, और थ्रेड का लाइफ़साइकल अनुमानित रहता है।