unbuffered streams

[Meep Meep]

selbstistdermeep schrieb:

Wenns nur 64 Bytes sind, warum nicht selber buffern?

Ansonsten ist das eine Sache für die Doku, denn es geht ja auch mit ungebufferten Streams, nur halt etwas langsamer (wobei mir die ganze Sache eher nach premature optimisation aussieht).

davon abgesehen das ich diese beiden woerter schon nicht mehr sehen kann, erklaer mir mal bitte was das damit zu tun hat

[otze]

selbstistdermeep schrieb:

(wobei mir die ganze Sache eher nach premature optimisation aussieht).

s gibt mehr als nur Dateistreams.

[hustbaer]

fuers parsen ist es wichtig das ich zu erst die ersten 64 bytes analysiere.
nach dem analysieren muss ich den offset-pointer aber wieder an den anfang des streams setzen.

Ich würde sagen dass du dir da ein grundlegendes Problem geschaffen hast, für dass es innerhalb der Iostreams Library keine Lösung gibt.

=> Selbst drum kümmern.

[Meep Meep]

also in die doku rein:
nur fuer gebufferte streams mit buffergroesse > 64 byte.

danke

Meep Meep

[SeppJ]

Meep Meep schrieb:

also in die doku rein:
nur fuer gebufferte streams mit buffergroesse > 64 byte.

Nee, das ist doch doof. Wieso nicht einfach die 64 Byte merken?

[Meep Meep]

SeppJ schrieb:

Meep Meep schrieb:

also in die doku rein:
nur fuer gebufferte streams mit buffergroesse > 64 byte.

Nee, das ist doch doof. Wieso nicht einfach die 64 Byte merken?

die ersten 64 bytes muessen analysiert werden damit ich weiß um welches format es sich handelt. kann ein reines textformat in ASCII, unicode usw sein oder aber auch ein binaeres format.
je nach dem um welches datei format es sich handelt, erzeuge ich einen stream dem ich den buffer uebergebe der dann die daten einliest.
darum spule ich auch den offset-pointer des streambuffers wieder an den anfang zurueck.
wenns nur filestreams waeren waere es kein problem. aber es kann auch ein socketstream oder sonst was sein, den ich nicht einfach schliessen und wieder oeffnen kann.
mir wuerde da auch nichts einfallen wie ich die 64 byte buffern und dann dem erzeugen stream mit dem streambuffer uebergeben koennte.

Meep Meep

[Ethon]

Lies doch einfach vom Inputstream in einen stringstream und arbeite mit dem weiter? Oder sind es zu viele Daten um sie im RAM zu halten? (Was ich wegen dem Socketstream mal bezweifle)

[out]

Ethon schrieb:

Lies doch einfach vom Inputstream in einen stringstream

Das macht Werner bestimmt traurig :p

[werni]

out schrieb:

Ethon schrieb:

Lies doch einfach vom Inputstream in einen stringstream

Das macht Werner bestimmt traurig :p

Der würde jetzt einen 300 Zeilen langen buffering_streambuf schreiben, der einen normalen streambuf wrappt, aber seekg für 64 Zeichen unterstützt (dann kommen nochmal 300 Zeilen für einen Ringbuffer hinzu).

Ist zwar alles möglich, aber ungemein umständlich und auch nicht wirklich performant (Performance spielt aber eh keine Rolle wenn der Input vom Netwerk kommt).

Pragmatische Lösung: Socketinput in eine Datei schreiben und die lesen (Posix bietet hier auch mmap an, Boost hat vielleicht auch sowas)
Schnelle Lösung (hier unangebracht): Der Read-Funktion einen zusätzlichen Parameter std::array<char,64> const& verpassen, der dann anfangs entsprchend behandelt wird.

[Ethon]

out schrieb:

Ethon schrieb:

Lies doch einfach vom Inputstream in einen stringstream

Das macht Werner bestimmt traurig :p

Warum? Ganz anderer Zusammenhang.

[Meep Meep]

wie gesagt es koennte ein socketstream sein. aber es gibt noch viele andere moeglichkeiten: mysqlstream, memorymappedfilestream, dbasestream, pipestream oder sonst was.

alleine von bestimmten socketstreams kann der stream gigabytes an daten liefern. der laeuft da stunden. in-memory geht da nichts. ansonsten haette ich erst garnicht streams genommen.
fuer einen systemlogger verwende ich z.b. einen pipestream (windows-pipes) der laeuft tagelang wo nur bestimmte ereignisse rausgenommen werden.

aber wie schon hustbaer schrib, giebt es dafuer keine loesung.

trotzdem danke an alle fuer ihren (vielleicht noch kommenden) input.

Meep Meep

[SeppJ]

werni schrieb:

out schrieb:

Ethon schrieb:

Lies doch einfach vom Inputstream in einen stringstream

Das macht Werner bestimmt traurig :p

Der würde jetzt einen 300 Zeilen langen buffering_streambuf schreiben, der einen normalen streambuf wrappt, aber seekg für 64 Zeichen unterstützt (dann kommen nochmal 300 Zeilen für einen Ringbuffer hinzu).

Ist zwar alles möglich, aber ungemein umständlich und auch nicht wirklich performant (Performance spielt aber eh keine Rolle wenn der Input vom Netwerk kommt).

Pragmatische Lösung: Socketinput in eine Datei schreiben und die lesen (Posix bietet hier auch mmap an, Boost hat vielleicht auch sowas)
Schnelle Lösung (hier unangebracht): Der Read-Funktion einen zusätzlichen Parameter std::array<char,64> const& verpassen, der dann anfangs entsprchend behandelt wird.

Übertreib mal nicht, so kompliziert ist das nun auch nicht gerade. Ganz einfaches Beispiel, bei dem ich aus Faulheit einen vector benutze, statt einem Ringpuffer:

#include <iostream>
#include <vector>

using namespace std;

class BufferedStreamBuf : public streambuf
{
  streambuf *buf;
  vector<int_type> buffer;
  streampos pos;
public:
  BufferedStreamBuf(streambuf* buf):  buf(buf), pos(0) { }
  BufferedStreamBuf(const BufferedStreamBuf&) = delete;
  BufferedStreamBuf& operator=(const BufferedStreamBuf&) = delete;
protected:
  virtual int_type underflow() 
  {
    if (pos == buffer.size())
      {
        int_type c = buf->sbumpc();
        buffer.push_back(c);
        return c;  
      }
    else if (pos < buffer.size())
      {
        return buffer[pos];
      }
    else
      {
        return EOF;
      }
  }
  virtual int_type uflow() 
  {
      if (pos >= buffer.size())
      {
        int_type c = buf->sbumpc();
        buffer.push_back(c);
        pos += 1;
        return c;  
      }
    else
      {
        return buffer[pos+=1];
      }
  }  
  virtual streampos seekoff(streamoff off, ios_base::seekdir way, ios_base::openmode which)
  {
    if (which == ios_base::in)
      {
        if (way == ios_base::beg and off >= 0)
          {
            while (off >= buffer.size())
              uflow();
            pos = off;             
          }
        if (way == ios_base::cur)
          {
            streampos new_pos = off + pos;
            while (new_pos >= buffer.size())
              uflow();
            pos = new_pos;
          }
       if (way == ios_base::end)
          {
            // Hatte keine Lust, die Standardlogik für diesen Fall nachzuschlagen.
            // Kann man später noch richtig machen. Erst einmal Fehler:
            return pos_type(off_type(-1));
          }        
      }
    return pos;
  }
 virtual streampos seekpos(streampos newp, ios_base::openmode which)
  {
    cerr << "Seekpos called.\n";
    if (which == ios_base::in)
      {
        while (newp >= buffer.size())
          uflow();
        pos = newp;             
      }
    return 0;
  }
 };

int main()
{
  BufferedStreamBuf buffered_cin_streambuf(cin.rdbuf());
  istream buffered_cin(&buffered_cin_streambuf);
  int i;
  buffered_cin >> i;
  cout << i << '\n';
  int j;
  buffered_cin.seekg(1, ios::beg);
  buffered_cin >> j;
  cout << j << '\n';
}

http://ideone.com/p4wpZu

Ich habe nicht alle Codepfade getestet. Könnte sein, dass ich mich irgendwo verhaspelt habe, das Beispiel funktioniert jedenfalls (man probiere es als Kontrast mit dem "normalen" cin, bei dem das seekg fehlschlägt). showmanyc, xsgetn und pbackfail könnte man noch implementieren, einfach weil man es bei einem buffered Stream sinnvoll implementieren kann, ebenso die Funktionen zum Setzen und Nutzen eines eigenen Zwischenspeichers. Sind aber nicht nötig, das 80 Zeilen Beispiel kann schon alles was man braucht.

[werni]

SeppJ schrieb:

showmanyc, xsgetn und pbackfail könnte man noch implementieren, einfach weil man es bei einem buffered Stream sinnvoll implementieren [soll]

Mach das noch, erbe von basic_streambuf, und buffere nur N Zeichen ... dann kommst du gut auf 300 Zeilen. So schlecht war meine Schätzung nicht

[SeppJ]

werni schrieb:

SeppJ schrieb:

showmanyc, xsgetn und pbackfail könnte man noch implementieren, einfach weil man es bei einem buffered Stream sinnvoll implementieren [soll]

Mach das noch, erbe von basic_streambuf, und buffere nur N Zeichen ... dann kommst du gut auf 300 Zeilen. So schlecht war meine Schätzung nicht

Da gibt es auch noch eine einfachere Lösung: Boost Iostreams. Hier im Forum reicht uns der Proof of Concept, in der richtigen Welt programmieren wir so etwas nicht komplett selber. Reicht schließlich, wenn das einmal jemand auf der Welt vernünftig implementiert hat, so dass es fortan alle benutzen können.

[Werner Salomon]

out schrieb:

Ethon schrieb:

Lies doch einfach vom Inputstream in einen stringstream

Das macht Werner bestimmt traurig :p

Nö - das geht ja noch. Bei Son'er Diskussion dagegen wusste ich wirklich nicht, ob ich lachen oder schreiend davon laufen soll.

werni schrieb:

Der würde jetzt einen 300 Zeilen langen buffering_streambuf schreiben, der einen normalen streambuf wrappt, aber seekg für 64 Zeichen unterstützt

.. ah! mein kleiner Bruder kennt mich.

werni schrieb:

(dann kommen nochmal 300 Zeilen für einen Ringbuffer hinzu).

Nur Anfänger brauchen dafür so viel Code

werni schrieb:

Ist zwar alles möglich, aber ungemein umständlich und auch nicht wirklich performant (Performance spielt aber eh keine Rolle wenn der Input vom Netwerk kommt).

Pragmatische Lösung: Socketinput in eine Datei schreiben und die lesen.

Es ist immer wieder erstaunlich, dass viele C++-IO per se für 'unperformat' halten - wohingegen mal eben das Wegschreiben in eine Datei und wieder lesen geht immer ganz schnell. werni Du stehst so gar nicht allein mit dieser Meinung, aber sie mutiert deshalb nicht vom Irrtum zur Wahrheit.

Ein deutlich unter 300 Code-Zeilen langer Wrapper-streambuf (mit Ringpuffer) der einen Buffer von N Byte garantiert:

#include <streambuf>
#include <cassert>

template< unsigned N, unsigned Reserve = 0 >
class BufferN : public std::streambuf
{
public:
    BufferN( std::streambuf* src )
        : src_( src )
        , p0_( 0 )
    {
        setg( buf_, buf_, buf_ );
    }
protected:
    virtual int_type underflow() override
    {
        assert( gptr() != 0 ); // garantiert der Konstruktor
        assert( gptr() == egptr() ); // ...sollte der Aufrufer garantieren
        const int_type m = src_->sbumpc();
        if( traits_type::eq_int_type( m, traits_type::eof() ) )
            return traits_type::eof();
        if( egptr() == end() )
        {
            traits_type::move( buf_, buf_+(Reserve+1), N-1 );
            setg( buf_, buf_+(N-1), buf_+N );
            p0_ += Reserve+1;
        }
        else
        {
            setg( buf_, egptr(), egptr()+1 );
        }
        *gptr() = traits_type::to_char_type( m );
        return m;
    }

    virtual pos_type seekoff( off_type off, std::ios_base::seekdir way, std::ios_base::openmode which = std::ios_base::in | std::ios_base::out) override
    {
        if( which != std::ios_base::in ) // nur input-Zeiger setzen erlaubt
            return pos_type( off_type(-1) );
        off_type local_off; // zeigt in den lokalen 'buf_'
        switch( way )
        {
        case std::ios_base::beg:
            local_off = off_type( off - p0_ );
            break;

        case std::ios_base::cur:
            local_off = off + (gptr() - eback());
            break;

        case std::ios_base::end: // not implemented
        default:
            return pos_type( off_type(-1) );
        }
        if( local_off < 0 || local_off > off_type(egptr() - eback()) )
            return pos_type( off_type(-1) ); // Fehler: local_off zeigt außerhalb der pending sequence
        setg( buf_, buf_+local_off, egptr() );
        return p0_ + local_off;
    }
    virtual pos_type seekpos( pos_type sp, std::ios_base::openmode which = std::ios_base::in | std::ios_base::out ) override
    {
        return seekoff( off_type(sp - (p0_ + off_type(gptr() - eback()))), std::ios_base::cur, which );
    }

private:
    char_type* end() { return buf_ + sizeof(buf_)/sizeof(*buf_); }
    std::streambuf* src_;
    char_type buf_[N+Reserve];
    pos_type p0_;
};

Sei der Input-Stream vom Typ std::istream oder einem Derivat, so klinkt man den Wrapper am einfachsten mit boost::io::ios_rdbuf_saver ein. Angenommen der Stream sei ' in ':

BufferN< 64 > sb( in.rdbuf() );
    boost::io::ios_rdbuf_saver svr( in, &sb ); // erfordert #include <boost/io/ios_state.hpp>

anschließend kann man mit pos0=in.tellg() sich die aktuelle Position merken, bis zu 64 Zeichen lesen und den Stream mit in. seekg (pos0) auf die alte Position zurücksetzen. putback und unget sind ebenso bis zu 64 Zeichen vor der Leseposition möglich.
Im Gegensatz zu SeppJs Lösung braucht es auch nur den N(+Reserve) großen Buffer und nicht die ganze Datei im Speicher. Theoretisch sollten auch Streams >4GByte (bzw. >std::numeric_limits< std::streamsize >::max()) gehandelt werden können - habe ich aber nicht ausprobiert.

Für die Performance-Junkies gibt es noch einen Template-Parameter Reserve . Setzt man den z.B. auf 10, so wird das move des Buffers nur bei jedem 10'ten Byte aufgerufen.

Gruß
Werner

[werni]

Werner Salomon schrieb:

Für die Performance-Junkies gibt es noch einen Template-Parameter Reserve . Setzt man den z.B. auf 10, so wird das move des Buffers nur bei jedem 10'ten Byte aufgerufen.

Ich meinte einen Circular Buffer, für den ist es nicht nötig, traits_type::move aufzurufen, da reicht es, den ältesten Character zu überschreiben (Reserve wird damit auch überflüssig).

Und was ich mit Performance meinte: Das Buffern wird eigentlich nur für die ersten 64 Bytes benötigt, beim restlichen Parsen wird das als Overhead mitgeschleppt. Hatte allerdings nicht an etwas wie boost::io::ios_rdbuf_saver gedacht, damit sollte das machbar sein (es wäre sogar ein Streambuf denkbar, der selbständig die Strategie wechselt).

Ich meinte damals, Streambuffs sind weder die schnellste denkbare Lösung (wegen dem Overhead: ein paar ifs und ein paar virtuelle Funktionsaufrufe pro char) noch die langsamste. Das in eine Datei schreiben wäre zwar nicht schneller, aber wenn ich 1h lang Daten runterlade kommt es auf eine Sekunde nachträgliches Parsen auch nicht mehr an.

[Werner Salomon]

werni schrieb:

Werner Salomon schrieb:

Für die Performance-Junkies gibt es noch einen Template-Parameter Reserve . Setzt man den z.B. auf 10, so wird das move des Buffers nur bei jedem 10'ten Byte aufgerufen.

Ich meinte einen Circular Buffer, für den ist es nicht nötig, traits_type::move aufzurufen, da reicht es, den ältesten Character zu überschreiben (Reserve wird damit auch überflüssig).

Ja - und ich meine den eben nicht, weil bei so kleinen Buffern (<100 Byte) ein Rotieren des Speichers (hier via move) nicht wirklich langsamer ist, als das Nachführen der Pointer - probiert's aus!
Mit circular_buffer müsste ich auch u.a. pbackfail neu implementieren, in meiner Lösung reicht das Default-Verhalten, da der Buffer die gesamte 'pending Sequenz' umfasst. Wäre also mehr Code, und genau das hast Du doch am Anfang kritisiert - oder?

werni schrieb:

Und was ich mit Performance meinte: Das Buffern wird eigentlich nur für die ersten 64 Bytes benötigt, beim restlichen Parsen wird das als Overhead mitgeschleppt. Hatte allerdings nicht an etwas wie boost::io::ios_rdbuf_saver gedacht, damit sollte das machbar sein (es wäre sogar ein Streambuf denkbar, der selbständig die Strategie wechselt).

Ich meinte damals, Streambuffs sind weder die schnellste denkbare Lösung (wegen dem Overhead: ein paar ifs und ein paar virtuelle Funktionsaufrufe pro char) noch die langsamste. Das in eine Datei schreiben wäre zwar nicht schneller, aber wenn ich 1h lang Daten runterlade kommt es auf eine Sekunde nachträgliches Parsen auch nicht mehr an.

.. da verstehe ich nicht, was Du damit sagen willst

Gruß
Werner

[hustbaer]

Anderer Vorschlag: nen Streambuffer der den Inhalt zweier (bzw. beliebig vieler) anderer Streambuffer zusammenhängt.
Müsste sich doch auch halbwegs einfach umsetzen lassen - schätz ich mal.

[werni]

Werner Salomon schrieb:

.. da verstehe ich nicht, was Du damit sagen willst

Wenn ich dein Streambuf mit einer ungenerischen Ad-Hoc-Lösung vergleiche ist bei deine ungemein langsamer:

void parse1(std::istreambuf_iterator<char> it, std::istreambuf_iterator<char> end,
            std::ostream& out, std::array<char, 64> const& arr)
{
  out << "Format 1\n";
  std::ostreambuf_iterator<char> o(out);
  std::copy(arr.begin(), arr.end(), o);
  std::copy(++it, end, o);
}

void parse2(std::istreambuf_iterator<char> it, std::istreambuf_iterator<char> end,
             std::ostream& out, std::array<char, 64> const& arr)
{
  out << "Format 2\n";
  std::ostreambuf_iterator<char> o(out);
  std::copy(arr.begin(), arr.end(), o);
  std::copy(++it, end, o);
}

void parse(std::istream& in, std::ostream& out)
{
  std::istreambuf_iterator<char> it(in), end;
  std::array<char, 64> save;
  std::copy_n(it, 64, save.begin());
  int magic = 17;
  for (int i=0; i<64; ++i)
    magic = 13*magic + save[i];
  if (magic%2)
    parse1(it, end, out, save);
  else
    parse2(in, end, out, save);
}

(das ist so in etwa, was ich von Meep² verstanden habe)

Vollständiger Vergleich: http://ideone.com/Tl3y8K
In meinen Messungen ergibt sich:

# Ad-Hoc-Hack:
real	0m0.168s
user	0m0.000s
sys 	0m0.304s

# Reserve=1
real	0m3.519s
user	0m3.352s
sys 	0m0.296s

# Reserve=1<<16
real    0m1.971s
user    0m1.840s
sys     0m0.252s

Dass deine Variante so langsam ist, hätte ich nicht gedacht, vielleicht habe ich da etwas falsch gemacht.

In diesem Fall wären -- wie auch hustbaer meint -- zwei Strategien angebracht, also ein "Streambuf [...] der selbständig die Strategie wechselt" (mein Post vorher).

[Werner Salomon]

werni schrieb:

Dass deine Variante so langsam ist, hätte ich nicht gedacht, vielleicht habe ich da etwas falsch gemacht.

Na ja - IMHO hast Du hier Äpfel mit Birnen verglichen. Das fängt schon damit an, dass Deine 'Ad-hoc-Lösung' keine Lösung ist. Zumindest hat der Anwender am Ende keinen Stream sondern ein Stück boost::array<64> plus einen Stream, wo er sich seine Sachen heraus pfrimeln darf. Stell Dir vor, da steht ab Position 62 bis Position 68 eine Zahl, die als solche geparst werden soll - das ist doch nur aufwendig. Diesen Aufwand hast Du nicht mit berechnet.
Ansonsten parst Du ja gar nicht, sondern machst nur ein Byte-Copy nach cout.

Ich habe mit selber so einen Test geschrieben, mit einem Buffer (im Memory!) mit 100kByte etwa zur Hälfte mit Text und zur anderen Hälfte mit Zahlen gefüllt. Wenn ich den BufferN<> zwischen istream und der eigentlichen Quelle setze, so verliert das Programm ca. 10% an Geschwindigkeit und absolut 1ms auf 100kByte. Dabei lese ich die Zahlen und überlese mit ignore die Texte. Wenn die Quelle jetzt eine Socket-Verbindung wäre, würde der relative Anteil noch weiter zurückgehen, da die Zeit für den Netz-Verkehr noch hinzukommt.

Und wem diese 1ms immer noch zu viel Zeitverlust sind, der kann einfach nach dem (zweiten und endgültigem) Lesen der ersten 64-plus-ein-paar-Bytes den Streambuf wieder zurück setzen. Man muss nur aufpassen einen Zeitpunkt zu wählen, bei dem beide Streambufs (Wrapper und Quelle) synchron sind - sich also keine ungelesenen Bytes im Wrapper befinden, die der Quelle bereits entzogen wurden. Das kann man prüfen, indem man die noch ausstehenden Bytes abfragt.
Etwa so:

if( in.rdbuf()->in_avail() == 0 ) // in := istream; aktuell mit dem Wrapper
            {
                in.rdbuf( &src ); // src := Quelle; streambuf umschalten
            }

und mit dieser Methode lässt sich überhaupt kein Performanceunterschied mehr feststellen.

Gruß
Werner