सी ++ इनपुट स्ट्रीम स्ट्रीम आउटपुट स्ट्रीम

Question

मैं जो करना चाहता हूं वह "पाइप" (प्रक्रियाओं के बीच एक पाइप की तरह) बनाना है, लेकिन उसी प्रोग्राम के भीतर सी ++ iostreams के बीच। मेरे पास एक ऐसा फ़ंक्शन है जिसके लिए एक इनपुट स्ट्रीम को तर्क के रूप में आवश्यक है, लेकिन मेरा डेटा आउटपुट स्ट्रीम से आ रहा है। तो के आउटपुट को std::istream के इनपुट में पाइप करने का एक मानक तरीका है?

Answer 1

आप std::streambuf बना सकते हैं जहां आउटपुट एक बफर और std::overflow() ब्लॉक हो जाता है जब बफर भर जाता है। दूसरी तरफ आपके पास इनपुट बफर होगा जो बफर खाली होने पर underflow() पर ब्लॉक करता है। जाहिर है, पढ़ना और लिखना दो अलग-अलग धागे में होगा।

मुश्किल व्यवसाय दो बफर को सिंक्रनाइज़ करने का तरीका है: धाराएं बफर तक पहुंचने के दौरान किसी सिंक्रनाइज़ेशन ऑपरेशंस का उपयोग नहीं करती हैं। केवल तभी जब किसी वर्चुअल फ़ंक्शंस को कॉल किया जाता है तो आप ऑपरेशन को रोक सकते हैं और सिंक्रनाइज़ेशन के साथ सौदा कर सकते हैं। दूसरी तरफ, बफर का उपयोग नहीं करना काफी अक्षम है। जिस तरह से मैं इस समस्या को संबोधित करता हूं वह अपेक्षाकृत छोटे आउटपुट बफर (उदाहरण के लिए 256 char एस) का उपयोग करके और इनपुट बफर में वर्णों के हस्तांतरण के लिए इस फ़ंक्शन का उपयोग करने के लिए sync() को ओवरराइड करता है। streambuf सिंक्रनाइज़ेशन के लिए एक म्यूटेक्स का उपयोग करेगा और आउटपुट पर एक पूर्ण इनपुट बफर पर इनपुट करने के लिए एक कंडीशन वैरिएबल और इनपुट पर एक खाली इनपुट बफर का उपयोग करेगा। स्वच्छ शटडाउन का समर्थन करने के लिए एक ध्वज स्थापित करने वाला फ़ंक्शन भी होना चाहिए जिसमें कोई इनपुट नहीं आ रहा है और सभी आउटपुट ऑपरेशन विफल हो जाएं।

वास्तविक कार्यान्वयन का निर्माण से पता चलता है कि दो बफर पर्याप्त नहीं हैं: इनपुट और आउटपुट बफर तक पहुंचने वाले थ्रेड सक्रिय हो सकते हैं जब संबंधित अन्य बफर ब्लॉक। इस प्रकार, एक तिहाई, मध्यवर्ती बफर की आवश्यकता है। उपर्युक्त योजना में इस छोटे बदलाव के साथ, नीचे कुछ कोड है (यह सुनिश्चित करने के लिए छोटे बफर का उपयोग करता है कि वास्तविक ओवरफ्लो और अंडरफ्लो हैं; वास्तविक उपयोग के लिए कम से कम इनपुट बफर शायद बड़ा होना चाहिए)।

// threadbuf.cpp              -*-C++-*- 
// ---------------------------------------------------------------------------- 
// Copyright (C) 2013 Dietmar Kuehl http://www.dietmar-kuehl.de   
//                  
// Permission is hereby granted, free of charge, to any person   
// obtaining a copy of this software and associated documentation  
// files (the "Software"), to deal in the Software without restriction, 
// including without limitation the rights to use, copy, modify,   
// merge, publish, distribute, sublicense, and/or sell copies of   
// the Software, and to permit persons to whom the Software is   
// furnished to do so, subject to the following conditions:    
//                  
// The above copyright notice and this permission notice shall be  
// included in all copies or substantial portions of the Software.  
//                  
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,  
// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES  
// OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND    
// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT   
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,   
// WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING   
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR   
// OTHER DEALINGS IN THE SOFTWARE. 
// ---------------------------------------------------------------------------- 


#include <algorithm> 
#include <condition_variable> 
#include <iostream> 
#include <mutex> 
#include <stdexcept> 
#include <streambuf> 
#include <string> 
#include <thread> 

// ---------------------------------------------------------------------------- 

class threadbuf 
    : public std::streambuf 
{ 
private: 
    typedef std::streambuf::traits_type traits_type; 
    typedef std::string::size_type  string_size_t; 

    std::mutex    d_mutex; 
    std::condition_variable d_condition; 
    std::string    d_out; 
    std::string    d_in; 
    std::string    d_tmp; 
    char*     d_current; 
    bool     d_closed; 

public: 
    threadbuf(string_size_t out_size = 16, string_size_t in_size = 64) 
     : d_out(std::max(string_size_t(1), out_size), ' ') 
     , d_in(std::max(string_size_t(1), in_size), ' ') 
     , d_tmp(std::max(string_size_t(1), in_size), ' ') 
     , d_current(&this->d_tmp[0]) 
     , d_closed(false) 
    { 
     this->setp(&this->d_out[0], &this->d_out[0] + this->d_out.size() - 1); 
     this->setg(&this->d_in[0], &this->d_in[0], &this->d_in[0]); 
    } 
    void close() 
    { 
     { 
      std::unique_lock<std::mutex> lock(this->d_mutex); 
      this->d_closed = true; 
      while (this->pbase() != this->pptr()) { 
       this->internal_sync(lock); 
      } 
     } 
     this->d_condition.notify_all(); 
    } 

private: 
    int_type underflow() 
    { 
     if (this->gptr() == this->egptr()) 
     { 
      std::unique_lock<std::mutex> lock(this->d_mutex); 
      while (&this->d_tmp[0] == this->d_current && !this->d_closed) { 
       this->d_condition.wait(lock); 
      } 
      if (&this->d_tmp[0] != this->d_current) { 
       std::streamsize size(this->d_current - &this->d_tmp[0]); 
       traits_type::copy(this->eback(), &this->d_tmp[0], 
            this->d_current - &this->d_tmp[0]); 
       this->setg(this->eback(), this->eback(), this->eback() + size); 
       this->d_current = &this->d_tmp[0]; 
       this->d_condition.notify_one(); 
      } 
     } 
     return this->gptr() == this->egptr() 
      ? traits_type::eof() 
      : traits_type::to_int_type(*this->gptr()); 
    } 
    int_type overflow(int_type c) 
    { 
     std::unique_lock<std::mutex> lock(this->d_mutex); 
     if (!traits_type::eq_int_type(c, traits_type::eof())) { 
      *this->pptr() = traits_type::to_char_type(c); 
      this->pbump(1); 
     } 
     return this->internal_sync(lock) 
      ? traits_type::eof() 
      : traits_type::not_eof(c); 
    } 
    int sync() 
    { 
     std::unique_lock<std::mutex> lock(this->d_mutex); 
     return this->internal_sync(lock); 
    } 
    int internal_sync(std::unique_lock<std::mutex>& lock) 
    { 
     char* end(&this->d_tmp[0] + this->d_tmp.size()); 
     while (this->d_current == end && !this->d_closed) { 
      this->d_condition.wait(lock); 
     } 
     if (this->d_current != end) 
     { 
      std::streamsize size(std::min(end - d_current, 
              this->pptr() - this->pbase())); 
      traits_type::copy(d_current, this->pbase(), size); 
      this->d_current += size; 
      std::streamsize remain((this->pptr() - this->pbase()) - size); 
      traits_type::move(this->pbase(), this->pptr(), remain); 
      this->setp(this->pbase(), this->epptr()); 
      this->pbump(remain); 
      this->d_condition.notify_one(); 
      return 0; 
     } 
     return traits_type::eof(); 
    } 
}; 

// ---------------------------------------------------------------------------- 

static void writer(std::ostream& out) 
{ 
    for (std::string line; std::getline(std::cin, line);) 
    { 
     out << "writer: '" << line << "'\n"; 
    } 
} 

// ---------------------------------------------------------------------------- 

static void reader(std::istream& in) 
{ 
    for (std::string line; std::getline(in, line);) 
    { 
     std::cout << "reader: '" << line << "'\n"; 
    } 
} 

// ---------------------------------------------------------------------------- 

int main() 
{ 
    try 
    { 
     threadbuf sbuf; 
     std::ostream out(&sbuf); 
     std::istream in(&sbuf); 

     std::thread write(&::writer, std::ref(out)); 
     std::thread read(&::reader, std::ref(in)); 

     write.join(); 
     sbuf.close(); 
     read.join(); 
    } 
    catch (std::exception const& ex) 
    { 
     std::cerr << "ERROR: " << ex.what() << "\n"; 
    } 
}