Line data    Source code

       1             : // The libMesh Finite Element Library.
       2             : // Copyright (C) 2002-2025 Benjamin S. Kirk, John W. Peterson, Roy H. Stogner
       3             : 
       4             : // This library is free software; you can redistribute it and/or
       5             : // modify it under the terms of the GNU Lesser General Public
       6             : // License as published by the Free Software Foundation; either
       7             : // version 2.1 of the License, or (at your option) any later version.
       8             : 
       9             : // This library is distributed in the hope that it will be useful,
      10             : // but WITHOUT ANY WARRANTY; without even the implied warranty of
      11             : // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
      12             : // Lesser General Public License for more details.
      13             : 
      14             : // You should have received a copy of the GNU Lesser General Public
      15             : // License along with this library; if not, write to the Free Software
      16             : // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
      17             : 
      18             : #include "libmesh/thread_buffered_syncbuf.h"
      19             : #include "libmesh/threads.h"
      20             : 
      21             : namespace libMesh
      22             : {
      23       19133 : ThreadBufferedSyncbuf::ThreadBufferedSyncbuf(std::streambuf & sink, bool flush_on_newline)
      24       19133 :   : _sink(sink), _flush_on_newline(flush_on_newline), _mu(std::make_unique<Threads::spin_mutex>())
      25             : {
      26       19133 : }
      27             : 
      28       35494 : ThreadBufferedSyncbuf::~ThreadBufferedSyncbuf() = default;
      29             : 
      30      164124 : ThreadBufferedSyncbuf::int_type ThreadBufferedSyncbuf::overflow(int_type ch)
      31             : {
      32      164124 :   if (traits_type::eq_int_type(ch, traits_type::eof()))
      33           0 :     return traits_type::not_eof(ch);
      34             : 
      35      164124 :   auto & t = this->thread_local_buffer();
      36      177635 :   t.buf.push_back(traits_type::to_char_type(ch));
      37             : 
      38      164124 :   if (_flush_on_newline && ch == '\n')
      39      164124 :     this->emit_from_thread_local_buffer(t.buf, /*force_flush=*/false);
      40             : 
      41       13630 :   return ch;
      42             : }
      43             : 
      44      574329 : std::streamsize ThreadBufferedSyncbuf::xsputn(const char * s, std::streamsize n)
      45             : {
      46      574329 :   auto & t = this->thread_local_buffer();
      47      574329 :   t.buf.append(s, static_cast<size_t>(n));
      48             : 
      49      574329 :   if (_flush_on_newline && !t.buf.empty() && t.buf.back() == '\n')
      50       18734 :     this->emit_from_thread_local_buffer(t.buf, /*force_flush=*/false);
      51             : 
      52      574329 :   return n;
      53             : }
      54             : 
      55      251387 : int ThreadBufferedSyncbuf::sync()
      56             : {
      57      251387 :   this->emit_from_thread_local_buffer(this->thread_local_buffer().buf, /*force_flush=*/true);
      58      251387 :   return 0;
      59             : }
      60             : 
      61      989840 : ThreadBufferedSyncbuf::ThreadLocalBuffer & ThreadBufferedSyncbuf::thread_local_buffer()
      62             : {
      63      989840 :   static thread_local ThreadLocalBuffer t(*this);
      64      989840 :   return t;
      65             : }
      66             : 
      67      434245 : void ThreadBufferedSyncbuf::emit_from_thread_local_buffer(std::string & b, bool force_flush)
      68             : {
      69      434245 :   if (b.empty())
      70             :     {
      71      186997 :       if (force_flush)
      72             :         {
      73       29218 :           Threads::spin_mutex::scoped_lock lk(*_mu);
      74      186997 :           _sink.pubsync();
      75             :         }
      76      186997 :       return;
      77             :     }
      78             : 
      79             :   {
      80       41802 :     Threads::spin_mutex::scoped_lock lk(*_mu);
      81      247248 :     _sink.sputn(b.data(), static_cast<std::streamsize>(b.size()));
      82      247248 :     if (force_flush)
      83       64390 :       _sink.pubsync();
      84             :   }
      85       20901 :   b.clear();
      86             : }
      87             : 
      88       16098 : ThreadBufferedSyncbuf::ThreadLocalBuffer::~ThreadLocalBuffer()
      89             : {
      90             :   // Runs at thread exit: commit any leftovers for this thread.
      91       16098 :   if (!buf.empty())
      92           0 :     _owner.emit_from_thread_local_buffer(buf, /*force_flush=*/false);
      93       16098 : }
      94             : 
      95             : }