2
\$\begingroup\$

This is a class from an implementation of SQL in C# I am writing. It fully buffers all reads and writes, and also logs changes to a log file so that updates are atomic ( all or nothing ). It implements the standard IO.Stream methods, but also has a "FastRead" method that allows direct access to the buffer to increase read performance, intended to speed up the rate at which a SQL base table can be scanned. All feedback appreciated ( and also feedback on the rest of the project, available at the github link above ).

namespace DBNS {

using IO = System.IO;
using G = System.Collections.Generic;

class FullyBufferedStream : IO.Stream
{
  // Implementation of IO.Stream in which all writes are buffered, and buffers are kept for every page access ( unless Rollback is called ).
  // Writes are recorded in the supplied log file, which ensures atomic updates.
  // Flush() must be called to write changes to the underlying stream, alternatively Rollback() may be called to discard changes.

  public FullyBufferedStream ( Log log, long fileId, IO.Stream f ) 
  { 
    Log = log;
    FileId = fileId;
    BaseStream = f; 
    Len = BaseStream.Length;
    Pos = 0;
    ReadAvail = 0;
    WriteAvail = 0;
  }

  readonly Log Log; // Log file to ensure atomic updates.
  readonly long FileId; // For Log file.
  readonly IO.Stream BaseStream; // Underlying stream.

  long Len; // File length
  long Pos; // Current position in the file

  readonly G.Dictionary<long,byte[]> Buffers = new G.Dictionary<long,byte[]>();
  readonly G.SortedSet<long> UnsavedPageNums = new G.SortedSet<long>();

  // Buffer size constants.
  const int BufferShift = 12; // Log base 2 of BufferSize.
  const int BufferSize = 1 << BufferShift;

  byte [] CurBuffer; // The current buffer.
  long CurBufferNum = -1;  // The page number of the current buffer.
  int CurIndex;      // Index into current buffer ( equal to Pos % BufferSize ).
  int WriteAvail;    // Number of bytes available for writing in CurBuffer ( from CurIndex ).
  int ReadAvail;     // Number of bytes available for reading in CurBuffer ( from CurIndex ).

  bool UnsavedAdded; // Current buffer has been added to UnsavedPageNums.

  public override long Seek( long to, System.IO.SeekOrigin how )
  { 
    long newpos;
    if ( how == System.IO.SeekOrigin.Begin )
      newpos = to;
    else if ( how == System.IO.SeekOrigin.End )
      newpos = Len + to;
    else // how == System.IO.SeekOrigin.Current
      newpos = Pos + to;
    if ( Pos != newpos )
    {
      Pos = newpos;
      WriteAvail = 0;
      ReadAvail = 0;
    }
    return newpos;
  }

  public override int Read( byte[] b, int off, int n )
  { 
    int request = n;
    while ( n > 0 )
    {
      int got = n > ReadAvail ? ReadAvail : n;
      if ( got > 0 )
      {
        for ( int i = 0; i < got; i += 1 ) b[ off + i ] = CurBuffer[ CurIndex + i ];
        off += got;
        n -= got;
        Pos += got;
        CurIndex += got;
        ReadAvail -= got;
      }
      else if ( Pos < Len ) DoSeek( true );
      else break;
    }
    return request - n;
  }

  // Instead of copying bytes, if possible Fastread returns the underlying buffer and an index into it.
  public byte[] FastRead( int n, out int ix )
  {
    if ( ReadAvail == 0 ) DoSeek( true );
    if ( ReadAvail >= n )
    {
      ix = CurIndex;
      CurIndex += n;
      Pos += n;
      ReadAvail -= n;
      return CurBuffer;
    }
    else
    {
      byte [] result = new byte[ n ];
      Read( result, 0, n );
      ix = 0;
      return result;
    }
  }  

  public override void Write( byte[] b, int off, int n )
  {
    Log.LogWrite( FileId, Pos, b, off, n );

    if ( Pos + n > Len ) Len = Pos + n;

    while ( n > 0 )
    {
      int got = n > WriteAvail ? WriteAvail : n;
      if ( got > 0 )
      {
        if ( !UnsavedAdded )
        {
          UnsavedPageNums.Add( CurBufferNum );
          UnsavedAdded = true;
        }
        for ( int i = 0; i < got; i += 1 ) CurBuffer[ CurIndex + i ] = b[ off + i ];
        off += got;
        n -= got;
        Pos += got;
        CurIndex += got;
        WriteAvail -= got;
      }
      else DoSeek( false );
    }
  }

  // Version of Write which checks the first byte written is zero.
  public bool Write( byte[] b, int off, int n, bool checkFirstByteZero )
  {
    Log.LogWrite( FileId, Pos, b, off, n );

    if ( Pos + n > Len ) Len = Pos + n;

    while ( n > 0 )
    {
      int got = n > WriteAvail ? WriteAvail : n;
      if ( got > 0 )
      {
        if ( checkFirstByteZero )
        {
          if ( CurBuffer[ CurIndex ] != 0 ) return false;
          checkFirstByteZero = false;
        }
        if ( !UnsavedAdded )
        {
          UnsavedPageNums.Add( CurBufferNum );
          UnsavedAdded = true;
        }
        for ( int i = 0; i < got; i += 1 ) CurBuffer[ CurIndex + i ] = b[ off + i ];
        off += got;
        n -= got;
        Pos += got;
        CurIndex += got;
        WriteAvail -= got;
      }
      else DoSeek( false );
    }
    return true;
  }

  public void Rollback()
  {
    Buffers.Clear();
    UnsavedPageNums.Clear();
    UnsavedAdded = false;
    ReadAvail = 0;
    WriteAvail = 0;   
    CurBufferNum = -1;
  }

  public override void Flush()
  {
    foreach ( long bufferNum in UnsavedPageNums )
    {
      byte [] b = GetBuffer( bufferNum );
      long pos = bufferNum << BufferShift;
      long n = Len - pos;
      if ( n > BufferSize ) n = BufferSize;
      if ( BaseStream.Position != pos ) BaseStream.Position = pos;
      BaseStream.Write( b, 0, (int)n );
    }
    UnsavedPageNums.Clear();
    UnsavedAdded = false;
    BaseStream.SetLength( Len );
    BaseStream.Flush();
  }

  public override void Close()
  {
    Rollback();
    BaseStream.Close();
  }

  public override void SetLength( long x )
  {
    Log.SetLength( FileId, x );
    Len = x;
    ReadAvail = 0;
    WriteAvail = 0;
  }

  public override bool CanRead { get{ return true; } }
  public override bool CanWrite { get{ return true; } }
  public override bool CanSeek { get{ return true; } }
  public override long Length { get{ return Len; } }
  public override long Position { get{ return Pos; } set{ Seek(value,0); } }

  void DoSeek( bool read )
  {
    if ( CurBufferNum != ( Pos >> BufferShift ) )
    {
      CurBufferNum = Pos >> BufferShift;
      CurBuffer = GetBuffer( CurBufferNum );
      UnsavedAdded = false;
    }

    CurIndex = (int) ( Pos & ( BufferSize - 1 ) );
    if ( read )
    {
      ReadAvail = BufferSize - CurIndex;
      if ( ReadAvail > Len - Pos ) ReadAvail = (int)( Len - Pos );
      WriteAvail = 0;
    }
    else
    {
      WriteAvail = BufferSize - CurIndex;
      ReadAvail = 0;
    }
  }

  byte [] GetBuffer( long bufferNum )
  {
    byte [] result;
    if ( Buffers.TryGetValue( bufferNum, out result ) ) return result;
    result = new byte[ BufferSize ];
    Buffers[ bufferNum ] = result;
    long pos = bufferNum << BufferShift;
    if ( BaseStream.Position != pos ) BaseStream.Position = pos;
    int i = 0;
    while ( i < BufferSize )
    {
      int got = BaseStream.Read( result, i, BufferSize - i );
      if ( got == 0 ) break;
      i += got;
    }
    return result;
  }

} // end class FullyBufferedStream

} // end namespace DBNS
\$\endgroup\$
4
  • \$\begingroup\$ When how == End in Seek(...), you set newpos = Len + to; - shouldn't that be newpos = Len - to;? Or else you seek beyond the length of the file? \$\endgroup\$ – user73941 Jun 13 '20 at 18:57
  • \$\begingroup\$ @HenrikHansen I had to go and check ( docs.microsoft.com/en-us/dotnet/api/system.io.seekorigin), but it says "The Seek methods take an offset parameter that is relative to the position specified by SeekOrigin" and the example uses negative offsets which further clarifies. So I think it's correct as is. \$\endgroup\$ – George Barwood Jun 13 '20 at 19:11
  • \$\begingroup\$ OK, I now realize that we are "inside" the stream, so you expect to to be negative. That makes sense, but I would maybe check the sign of to before adding it. \$\endgroup\$ – user73941 Jun 14 '20 at 4:19
  • 1
    \$\begingroup\$ It's legal ( and somewhat normal ) to seek beyond the end of the stream and then write, so positive offsets from End are not wrong - that said, I don't use Seek at all, let alone the End and Current options, instead assigning Position seems simpler. Which suggests a change : I think I will make that the "core" operation, with Seek setting position rather than Position calling Seek. \$\endgroup\$ – George Barwood Jun 14 '20 at 5:08
1
\$\begingroup\$

I discovered what I think is a flaw - in Flush() I call simply Flush() on the underlying stream, however I think this needs to be Flush(true) to ensure that all the changes are written before the log file is reset. I made the same change in Log.cs.

This means changing the type of BaseStream to FileStream, as Flush(true) is specific to FileStream.

The primitive I want is some kind of "fence" to say "write operations (all FileStreams) must not be re-ordered over this fence (either way)." However this doesn't seem to be available unless I have missed something - or perhaps plain Flush() has this effect, but if so, it doesn't seem to be documented.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.