v0.27

Author: kolomenkin

LineReader.cpp

@@ -356,3 +356,155 @@ std::optional<std::string_view> CMappingLineReader::GetNextLine()
     assert(foundLineLength > 0 && "result should contain at least LF char");
     return result;
 }
+
+//////////////////////////////////////////////////////////////////////////
+//////////////////////////////////////////////////////////////////////////
+//////////////////////////////////////////////////////////////////////////
+
+CLockFreeLineReader::CLockFreeLineReader()
+{
+    this->_buffer1.Allocate(ReadBufferSize);
+    if (this->_buffer1.ptr != nullptr)
+    {
+        this->_buffer2.Allocate(ReadBufferSize);
+    }
+}
+
+bool CLockFreeLineReader::Open(const wchar_t* const filename)
+{
+    if (this->_buffer1.ptr == nullptr || filename == nullptr)
+    {
+        return false;
+    }
+    assert(this->_buffer2.ptr != nullptr);
+    this->Close();
+
+    const bool bAsyncMode = false;
+    const bool succeeded = this->_file.Open(filename, bAsyncMode);
+    if (!succeeded)
+    {
+        return false;
+    }
+
+    this->_bufferData = std::string_view(this->_buffer1.ptr, 0);
+    this->_firstBufferIsActive = true;
+
+    const bool initLockFreeOk = this->_file.LockFreecInit();
+    if (!initLockFreeOk)
+    {
+        this->_file.Close();
+        return false;
+    }
+
+    const bool readStartOk = this->_file.LockFreeReadStart(this->_buffer2.ptr + ReadBufferOffset, ReadChunkSize);
+    if (!readStartOk)
+    {
+        this->_file.LockFreecClean();
+        this->_file.Close();
+        return false;
+    }
+
+    return true;
+}
+
+void CLockFreeLineReader::Close()
+{
+    this->_file.LockFreecClean();
+    this->_file.Close();
+}
+
+__declspec(noinline) // noinline is added to help CPU profiling in release version
+std::optional<std::string_view> CLockFreeLineReader::GetNextLine()
+{
+    if (this->_buffer1.ptr == nullptr)
+    {
+        return {};
+    }
+    assert(this->_buffer2.ptr != nullptr);
+
+    // Find EOL:
+    size_t eolOffset = this->_bufferData.find('\n');
+
+    if (eolOffset == this->_bufferData.npos)
+    {
+        // EOL was not found. Make a choice between last line case and reading additional data from functor.
+
+        if (this->_bufferData.size() > MaxLogLineLength)
+        {
+            // Incomplete line is already too long
+            return {};
+        }
+
+        CCharBuffer& currentBuffer = this->_firstBufferIsActive ? this->_buffer1 : this->_buffer2;
+        CCharBuffer& nextBuffer = this->_firstBufferIsActive ? this->_buffer2 : this->_buffer1;
+
+        const size_t prefixLength = this->_bufferData.size();
+        assert(prefixLength <= MaxLogLineLength && "the rest of buffer is too big for moving to beginning");
+        char* const newDataBufferPtr = nextBuffer.ptr + ReadBufferOffset - prefixLength;
+
+        // don't need memmove since the whole high level algorithm will fail if buffers overlap
+        memcpy(newDataBufferPtr, this->_bufferData.data(), prefixLength);
+
+        size_t readBytes = 0;
+        const bool readCompleteOk = this->_file.LockFreeReadWait(readBytes);
+        if (!readCompleteOk)
+        {
+            // Previous reading failed
+            return {};
+        }
+
+        // Read missing data:
+        const bool readOk = this->_file.LockFreeReadStart(currentBuffer.ptr + ReadBufferOffset, ReadChunkSize);
+        if (!readOk)
+        {
+            // New reading failed
+            return {};
+        }
+
+        this->_bufferData = { newDataBufferPtr, prefixLength + readBytes };
+        this->_firstBufferIsActive = !this->_firstBufferIsActive;
+
+        if (this->_bufferData.empty())
+        {
+            assert(readBytes == 0);
+            // The very last line without LF is not counted
+            return {};
+        }
+
+        // Search EOL again after reading additional data:
+        // I expect we read either ReadChunkSize bytes or we read the data chunk in file.
+        eolOffset = this->_bufferData.find('\n', prefixLength);
+        if (eolOffset == this->_bufferData.npos)
+        {
+            if (this->_bufferData.size() > MaxLogLineLength)
+            {
+                // Incomplete line is too long
+                return {};
+            }
+
+            // Found last line after reading missing data
+            const std::string_view result = this->_bufferData;
+            this->_bufferData = { this->_buffer1.ptr, 0 };
+            assert(!result.empty() && "last line without LF should be not empty");
+            return result;
+        }
+    }
+
+    const size_t foundLineLength = eolOffset + 1;
+
+    if (foundLineLength > MaxLogLineLength)
+    {
+        // Line is too long
+        return {};
+    }
+
+    const std::string_view result = this->_bufferData.substr(0, foundLineLength);
+    this->_bufferData.remove_prefix(foundLineLength);
+
+    assert(foundLineLength > 0 && "result should contain at least LF char");
+    return result;
+}
+
+//////////////////////////////////////////////////////////////////////////
+//////////////////////////////////////////////////////////////////////////
+//////////////////////////////////////////////////////////////////////////

LineReader.h

@@ -70,3 +70,29 @@ class CMappingLineReader
     bool             _mappedToMemory = false;
     std::string_view _bufferData; // filled part of the current buffer
 };
+
+//////////////////////////////////////////////////////////////////////////
+
+class CLockFreeLineReader
+{
+public:
+    CLockFreeLineReader();
+
+    bool Open(const wchar_t* const filename);
+    void Close();
+
+    // request next matching line; line may contain '\0' and may end with '\n'; return false on error or EOF
+    // returned line is never empty (it contains at least one '\n' or any other character).
+    std::optional<std::string_view> GetNextLine();
+
+protected:
+    CScanFile        _file;
+    // Buffer structure: [    rest_of_previousline|data_read_from_file  ]
+    //                   [ len = MaxLogLineLength | len = ReadChunkSize ]
+    bool             _firstBufferIsActive = true;
+    CCharBuffer      _buffer1;
+    CCharBuffer      _buffer2;
+    std::string_view _bufferData; // filled part of the current buffer
+};
+
+//////////////////////////////////////////////////////////////////////////

LogReader.h

@@ -47,14 +47,18 @@ class CLogReader final
     }
 
 protected:
-#if 1
+#if 0
 #if 0
     CSyncLineReader    _lineReader;
 #else
-    CAsyncLineReader   _lineReader;
+    CMappingLineReader _lineReader;
 #endif
 #else
-    CMappingLineReader _lineReader;
+#if 0
+    CAsyncLineReader    _lineReader;
+#else
+    CLockFreeLineReader _lineReader;
+#endif
 #endif
     CCharBuffer        _pattern;
     CFnMatch           _lineMatcher;

README.md

@@ -75,6 +75,6 @@
 
 Как и в условии задачи, основное консольное приложение собрано с отключенными исключениями и заодно без RTTI.
 
-Я притянул часть STL на мой страх и риск. Ту, часть, которая не требует исключений и работает без лишних накладных расходов.
+Я притянул часть STL на мой страх и риск. Ту часть, которая не требует исключений и работает без лишних накладных расходов.
 
 ---

ScanFile.cpp

@@ -18,7 +18,6 @@ bool CScanFile::Open(const wchar_t* const filename, const bool asyncMode)
     {
         return false;
     }
-
     assert(this->_hEvent == nullptr);
 
     const DWORD dwDesiredAccess = FILE_READ_DATA | FILE_READ_ATTRIBUTES; // minimal required rights
@@ -38,6 +37,7 @@ bool CScanFile::Open(const wchar_t* const filename, const bool asyncMode)
         return false;
     }
 
+    // Init data for async IO:
     this->_hEvent = CreateEventW(nullptr, TRUE, FALSE, nullptr);
 
     if (this->_hEvent == nullptr)
@@ -56,6 +56,8 @@ bool CScanFile::Open(const wchar_t* const filename, const bool asyncMode)
 
 void CScanFile::Close()
 {
+    this->LockFreecClean();
+
     if (this->_pViewOfFile != nullptr)
     {
         UnmapViewOfFile(this->_pViewOfFile);
@@ -129,6 +131,8 @@ std::optional<std::string_view> CScanFile::MapToMemory()
     return std::string_view(static_cast<const char*>(this->_pViewOfFile), fileSizeAsSizeT);
 }
 
+//////////////////////////////////////////////////////////////////////////
+
 __declspec(noinline) // noinline is added to help CPU profiling in release version
 bool CScanFile::Read(char* const buffer, const size_t bufferLength, size_t& readBytes)
 {
@@ -152,6 +156,8 @@ bool CScanFile::Read(char* const buffer, const size_t bufferLength, size_t& read
     return !!succeeded;
 }
 
+//////////////////////////////////////////////////////////////////////////
+
 __declspec(noinline) // noinline is added to help CPU profiling in release version
 bool CScanFile::AsyncReadStart(char* const buffer, const size_t bufferLength)
 {
@@ -210,3 +216,132 @@ bool CScanFile::AsyncReadWait(size_t& readBytes)
 
     return true;
 }
+
+//////////////////////////////////////////////////////////////////////////
+
+bool CScanFile::LockFreecInit()
+{
+    if (this->_pThread != nullptr)
+    {
+        return false;
+    }
+
+    this->_threadFinishSignal = false;
+    this->_threadOperationReadStartSignal = false;
+    this->_threadOperationReadCompletedSignal = false;
+    {
+        std::lock_guard<std::mutex> lock(this->_protectThreadData);
+        this->_pThreadReadBuffer = nullptr;
+        this->_threadReadBufferSize = 0;
+        this->_threadActuallyReadBytes = 0;
+        this->_threadReadSucceeded = false;
+    }
+
+    this->_pThread = std::make_unique<std::thread>(&CScanFile::LockFreeThread, this);
+
+    return true;
+}
+
+void CScanFile::LockFreecClean()
+{
+    if (this->_pThread != nullptr)
+    {
+        this->_threadFinishSignal = true;
+        this->_pThread->join();
+        this->_pThread.reset();
+    }
+}
+
+__declspec(noinline) // noinline is added to help CPU profiling in release version
+bool CScanFile::LockFreeReadStart(char* const buffer, const size_t bufferLength)
+{
+    if (this->_pThread == nullptr || this->_threadOperationInProgress)
+    {
+        return false;
+    }
+
+    this->_threadOperationInProgress = true;
+
+    this->_threadOperationReadCompletedSignal = false;
+
+    {
+        std::lock_guard<std::mutex> lock(this->_protectThreadData);
+        this->_pThreadReadBuffer = buffer;
+        this->_threadReadBufferSize = bufferLength;
+        this->_threadActuallyReadBytes = 0;
+        this->_threadReadSucceeded = false;
+    }
+
+    this->_threadOperationReadStartSignal = true;
+
+    return true;
+}
+
+__declspec(noinline) // noinline is added to help CPU profiling in release version
+bool CScanFile::LockFreeReadWait(size_t& readBytes)
+{
+    if (this->_pThread == nullptr || !this->_threadOperationInProgress)
+    {
+        return false;
+    }
+
+    this->_threadOperationInProgress = false;
+
+    while (!this->_threadOperationReadCompletedSignal)
+    {
+        // nothing
+    }
+
+    {
+        std::lock_guard<std::mutex> lock(this->_protectThreadData);
+        readBytes = this->_threadActuallyReadBytes;
+        if (!this->_threadReadSucceeded)
+        {
+            return false;
+        }
+    }
+
+    return true;
+}
+
+void CScanFile::LockFreeThread()
+{
+    while (true)
+    {
+        if (this->_threadFinishSignal)
+        {
+            // thread exit signal is caught
+            break;
+        }
+
+        if (!this->_threadOperationReadStartSignal)
+        {
+            // nothing to do
+            continue;
+        }
+
+        this->_threadOperationReadStartSignal = false;
+
+        char* buffer = nullptr;
+        size_t bufferSize = 0;
+
+        {
+            std::lock_guard<std::mutex> lock(this->_protectThreadData);
+            buffer = this->_pThreadReadBuffer;
+            bufferSize = this->_threadReadBufferSize;
+        }
+
+        size_t readBytes = 0;
+        const bool readOk = this->Read(buffer, bufferSize, readBytes);
+
+        {
+            std::lock_guard<std::mutex> lock(this->_protectThreadData);
+            this->_threadActuallyReadBytes = readBytes;
+            this->_threadReadSucceeded = readOk;
+        }
+
+        this->_threadOperationReadCompletedSignal = true;
+    }
+}
+
+//////////////////////////////////////////////////////////////////////////