Buffer.cc

/*
 * Copyright (c) 2008-2014, Pedigree Developers
 *
 * Please see the CONTRIB file in the root of the source tree for a full
 * list of contributors.
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */
 
#include "pedigree/kernel/LockGuard.h"
#include "pedigree/kernel/Log.h"
#include "pedigree/kernel/utilities/Buffer.h"
#include "pedigree/kernel/utilities/assert.h"
#include "pedigree/kernel/utilities/utility.h"
 
#if THREADS
#include "pedigree/kernel/process/Thread.h"
#include "pedigree/kernel/processor/Processor.h"
#include "pedigree/kernel/processor/ProcessorInformation.h"
 
namespace {
void preserveConditionInterruption(ConditionVariable::Error error) {
  if (error == ConditionVariable::Interrupted) {
    Processor::information().getCurrentThread()->setInterruptionReason(Thread::InterruptedBySignal);
  }
}
}  // namespace
#endif
 
template <class T, bool allowShortOperation>
Buffer<T, allowShortOperation>::Buffer(size_t bufferSize)
    : m_BufferSize(bufferSize),
      m_DataSize(0),
      m_ReadableGeneration(0),
      m_WritableGeneration(0),
      m_Lock(),
      m_WriteCondition(),
      m_ReadCondition(),
      m_DrainCondition(),
      m_Segments(),
      m_Monitors(),
      m_bCanRead(true),
      m_bCanWrite(true),
      m_bClosing(false),
      m_ActiveOperations(0) {}
 
template <class T, bool allowShortOperation>
Buffer<T, allowShortOperation>::~Buffer() {
  TerminationDeferral terminationDeferral;
  m_Lock.acquire();
  m_bClosing = true;
  m_bCanRead = false;
  m_bCanWrite = false;
  m_ReadCondition.broadcast();
  m_WriteCondition.broadcast();
 
  while (m_ActiveOperations) {
    m_DrainCondition.waitForCompletion(m_Lock);
  }
 
  for (auto pSegment : m_Segments) {
    delete pSegment;
  }
  m_Segments.clear();
  m_DataSize = 0;
 
  m_Monitors.clear();
  m_Lock.release();
}
 
template <class T, bool allowShortOperation>
bool Buffer<T, allowShortOperation>::beginOperation() {
  LockGuard<Mutex> guard(m_Lock);
  if (m_bClosing) {
    return false;
  }
 
  ++m_ActiveOperations;
  return true;
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::endOperation() {
  LockGuard<Mutex> guard(m_Lock);
  assert(m_ActiveOperations);
  --m_ActiveOperations;
  if (m_bClosing && !m_ActiveOperations) {
    m_DrainCondition.signal();
  }
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::write(const T* buffer, size_t count, bool block) {
  ActiveOperation operation(*this);
  if (!operation) {
    return 0;
  }
 
  if (!block) {
    if (!m_Lock.tryAcquire()) {
      // can't unlock buffer for writing
      return 0;
    }
  } else {
    // can block!
    m_Lock.acquire();
  }
 
  size_t countSoFar = 0;
  while (true) {
    // Can we write?
    if (!m_bCanWrite) {
      // No! Maybe not anymore, so return what we've written so far.
      break;
    }
 
    // Do we have space?
    size_t bytesAvailable = m_BufferSize - m_DataSize;
    if (!bytesAvailable) {
      if (!block) {
        // Cannot block!
        break;
      }
 
      // Can any reader get us out of this situation?
      if (!m_bCanRead) {
        // No. Return what we've written so far.
        break;
      }
 
      // No, we need to wait.
      ConditionVariable::Error error = ConditionVariable::NoError;
      if (!m_WriteCondition.wait(m_Lock, error)) {
#if THREADS
        preserveConditionInterruption(error);
#endif
        if (ConditionVariable::mutexAcquired(error)) {
          m_Lock.release();
        }
        return countSoFar;
      }
      continue;
    }
 
    // Yes, we have room.
    size_t totalCount = bytesAvailable;
    if (totalCount > count) {
      totalCount = count;
    }
 
    // If we're allowed to just give up if we don't have enough bytes, do
    // so (e.g. for TCP buffers and the like).
    if (allowShortOperation && (totalCount > bytesAvailable)) {
      totalCount = bytesAvailable;
      count = bytesAvailable;
      if (!totalCount) {
        break;
      }
    }
 
    const size_t numberCopied = writeLocked(buffer, totalCount);
    countSoFar += numberCopied;
    buffer += numberCopied;
    count -= numberCopied;
 
    // Wake up a reader that was waiting for data.
    // We do this here rather than after we finish as we may need to block
    // again (e.g. if we're writing more than the size of the buffer), and
    // a reader is needed to unblock that.
    if (numberCopied) {
      m_ReadCondition.signal();
    }
 
    if (!count) {
      // Complete.
      break;
    }
  }
 
  m_Lock.release();
 
  if (countSoFar) {
    // We've updated the buffer, so send events.
    notifyMonitors();
  }
 
  return countSoFar;
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::writeAvailable(const T* buffer, size_t count, bool atomic) {
  ActiveOperation operation(*this);
  if (!operation) {
    return 0;
  }
 
  m_Lock.acquire();
  size_t written = 0;
  if (m_bCanRead && m_bCanWrite) {
    const size_t available = m_BufferSize - m_DataSize;
    if (!atomic || count <= available) {
      written = writeLocked(buffer, count < available ? count : available);
      if (written) {
        m_ReadCondition.signal();
      }
    }
  }
  m_Lock.release();
 
  if (written) {
    notifyMonitors();
  }
  return written;
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::writeAtomic(const T* buffer, size_t count, bool block) {
  ActiveOperation operation(*this);
  if (!operation || count > m_BufferSize) {
    return 0;
  }
 
  if (!block) {
    if (!m_Lock.tryAcquire()) {
      return 0;
    }
  } else {
    m_Lock.acquire();
  }
 
  size_t written = 0;
  while (m_bCanWrite && m_bCanRead) {
    if (count <= (m_BufferSize - m_DataSize)) {
      written = writeLocked(buffer, count);
      if (written) {
        m_ReadCondition.signal();
      }
      break;
    }
 
    if (!block || !m_bCanRead) {
      break;
    }
 
    ConditionVariable::Error error = ConditionVariable::NoError;
    if (!m_WriteCondition.wait(m_Lock, error)) {
#if THREADS
      preserveConditionInterruption(error);
#endif
      if (ConditionVariable::mutexAcquired(error)) {
        m_Lock.release();
      }
      return 0;
    }
  }
 
  m_Lock.release();
  if (written) {
    notifyMonitors();
  }
  return written;
}
 
template <class T, bool allowShortOperation>
bool Buffer<T, allowShortOperation>::tryWrite(const T* buffer, size_t count) {
  TerminationDeferral terminationDeferral;
  if (!m_Lock.tryAcquire()) {
    return false;
  }
 
  if (m_bClosing || !m_bCanWrite || count > (m_BufferSize - m_DataSize)) {
    m_Lock.release();
    return false;
  }
 
  const size_t numberCopied = writeLocked(buffer, count);
  if (numberCopied) {
    m_ReadCondition.signal();
    notifyMonitorsLocked();
  }
  m_Lock.release();
  return numberCopied == count;
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::read(T* buffer, size_t count, bool block) {
  ActiveOperation operation(*this);
  if (!operation) {
    return 0;
  }
 
  // Nonblocking reads must still wait for lock ownership, as beginOperation
  // already does. Contention is not an empty buffer (or EOF to the caller).
  m_Lock.acquire();
 
  size_t countSoFar = 0;
  while (true) {
    // Can we read?
    if (!m_bCanRead) {
      // No! Maybe not anymore, so return what we've read so far.
      break;
    }
 
    // Do we have anything to read?
    if (!m_DataSize) {
      if (!block) {
        // Cannot block!
        break;
      }
 
      // Can any writer get us out of this situation?
      if (!m_bCanWrite) {
        // No. Return what we've read so far.
        break;
      }
 
      // No, we need to wait.
      ConditionVariable::Error error = ConditionVariable::NoError;
      if (!m_ReadCondition.wait(m_Lock, error)) {
#if THREADS
        preserveConditionInterruption(error);
#endif
        if (ConditionVariable::mutexAcquired(error)) {
          m_Lock.release();
        }
        return countSoFar;
      }
      continue;
    }
 
    // Yes, we have room.
    size_t totalCount = count;
    if (totalCount > m_DataSize) {
      totalCount = m_DataSize;
    }
 
    size_t numberCopied = 0;
    while (m_Segments.count() && numberCopied < totalCount) {
      // Grab the first segment and read it.
      Segment* pSegment = m_Segments.popFront();
      size_t countToRead = pSegment->size - pSegment->reader;
      if ((numberCopied + countToRead) > totalCount) {
        countToRead = totalCount - numberCopied;
      }
 
      // Copy.
      pedigree_std::copy(buffer, &pSegment->data[pSegment->reader], countToRead);
      pSegment->reader += countToRead;
 
      // Do we need to re-add it?
      if (pSegment->reader < pSegment->size) {
        m_Segments.pushFront(pSegment);
      } else {
        delete pSegment;
      }
 
      numberCopied += countToRead;
      buffer += countToRead;
    }
 
    const bool wasFull = m_DataSize >= m_BufferSize;
    m_DataSize -= numberCopied;
    if (wasFull && m_DataSize < m_BufferSize) {
      m_WritableGeneration += 1;
    }
    countSoFar += numberCopied;
    count -= numberCopied;
 
    // We read some bytes so writers may be able to continue, which may be
    // needed to unblock us if we loop back around and block.
    if (numberCopied) {
      // Wake up a writer that was waiting for space to write.
      m_WriteCondition.signal();
    }
 
    if (!count) {
      break;
    }
 
    // Once we've read at least some bytes, don't block - just return what
    // we've read so far if we loop back around and have no data.
    block = false;
  }
 
  m_Lock.release();
 
  if (countSoFar) {
    // We've updated the buffer, so send events.
    notifyMonitors();
  }
 
  return countSoFar;
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::disableWrites() {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
  LockGuard<Mutex> guard(m_Lock);
  m_bCanWrite = false;
 
  // Readers may be waiting for data, while writers may already be waiting
  // for space. Both predicates have changed permanently for this direction.
  m_ReadCondition.broadcast();
  m_WriteCondition.broadcast();
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::disableReads() {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
  LockGuard<Mutex> guard(m_Lock);
  m_bCanRead = false;
 
  // Writers may be waiting for space, while readers may already be waiting
  // for data. Both predicates have changed permanently for this direction.
  m_ReadCondition.broadcast();
  m_WriteCondition.broadcast();
}
 
template <class T, bool allowShortOperation>
bool Buffer<T, allowShortOperation>::enableWrites() {
  ActiveOperation operation(*this);
  if (!operation) {
    return false;
  }
 
  LockGuard<Mutex> guard(m_Lock);
  bool previous = m_bCanWrite;
  m_bCanWrite = true;
  if (!previous && m_DataSize < m_BufferSize) {
    m_WritableGeneration += 1;
  }
  return previous;
}
 
template <class T, bool allowShortOperation>
bool Buffer<T, allowShortOperation>::enableReads() {
  ActiveOperation operation(*this);
  if (!operation) {
    return false;
  }
 
  LockGuard<Mutex> guard(m_Lock);
  bool previous = m_bCanRead;
  m_bCanRead = true;
  if (!previous && m_DataSize) {
    m_ReadableGeneration += 1;
  }
  return previous;
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::getDataSize() {
  ActiveOperation operation(*this);
  if (!operation) {
    return 0;
  }
 
  LockGuard<Mutex> guard(m_Lock);
  return m_DataSize;
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::getSize() {
  ActiveOperation operation(*this);
  if (!operation) {
    return 0;
  }
 
  return m_BufferSize;
}
 
template <class T, bool allowShortOperation>
bool Buffer<T, allowShortOperation>::canWrite(bool block) {
  ActiveOperation operation(*this);
  if (!operation) {
    return false;
  }
 
  LockGuard<Mutex> guard(m_Lock);
 
  if (!block) {
    return m_bCanWrite && (m_DataSize < m_BufferSize);
  }
 
  if (!m_bCanWrite) {
    return false;
  }
 
  // A full buffer can only become writable while reads remain possible.
  while (m_bCanWrite && m_bCanRead && m_DataSize >= m_BufferSize) {
    ConditionVariable::Error error = ConditionVariable::NoError;
    if (!m_WriteCondition.wait(m_Lock, error)) {
#if THREADS
      preserveConditionInterruption(error);
#endif
      if (!ConditionVariable::mutexAcquired(error)) {
        guard.disown();
      }
      return false;
    }
  }
 
  return m_bCanWrite && m_DataSize < m_BufferSize;
}
 
template <class T, bool allowShortOperation>
bool Buffer<T, allowShortOperation>::canRead(bool block) {
  ActiveOperation operation(*this);
  if (!operation) {
    return false;
  }
 
  LockGuard<Mutex> guard(m_Lock);
 
  if (!block) {
    return m_bCanRead && (m_DataSize > 0);
  }
 
  if (!m_bCanRead) {
    return false;
  }
 
  // An empty buffer can only become readable while writes remain possible.
  while (m_bCanRead && m_bCanWrite && !m_DataSize) {
    ConditionVariable::Error error = ConditionVariable::NoError;
    if (!m_ReadCondition.wait(m_Lock, error)) {
#if THREADS
      preserveConditionInterruption(error);
#endif
      if (!ConditionVariable::mutexAcquired(error)) {
        guard.disown();
      }
      return false;
    }
  }
 
  return m_bCanRead && m_DataSize > 0;
}
 
template <class T, bool allowShortOperation>
uint64_t Buffer<T, allowShortOperation>::readableGeneration() const {
  return m_ReadableGeneration.value();
}
 
template <class T, bool allowShortOperation>
uint64_t Buffer<T, allowShortOperation>::writableGeneration() const {
  return m_WritableGeneration.value();
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::wipe() {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
  LockGuard<Mutex> guard(m_Lock);
 
  const bool wasFull = m_DataSize >= m_BufferSize;
 
  // Wipe out every segment we own.
  for (auto pSegment : m_Segments) {
    delete pSegment;
  }
  m_Segments.clear();
  m_DataSize = 0;
  if (wasFull && m_BufferSize) {
    m_WritableGeneration += 1;
  }
 
  // Notify writers that might have been waiting for space.
  m_WriteCondition.signal();
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::monitor(Thread* pThread, Event* pEvent) {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
#if THREADS
  LockGuard<Mutex> guard(m_Lock);
  m_Monitors.add(pThread, pEvent);
#endif
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::monitor(Semaphore* pSemaphore) {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
#if THREADS
  LockGuard<Mutex> guard(m_Lock);
  m_Monitors.add(pSemaphore);
#endif
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::cullMonitorTargets(Thread* pThread) {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
#if THREADS
  LockGuard<Mutex> guard(m_Lock);
  m_Monitors.cull(pThread);
#endif
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::cullMonitorTargets(Semaphore* pSemaphore) {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
#if THREADS
  LockGuard<Mutex> guard(m_Lock);
  m_Monitors.cull(pSemaphore);
#endif
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::cullMonitorTargets(Event* pEvent) {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
#if THREADS
  LockGuard<Mutex> guard(m_Lock);
  m_Monitors.cull(pEvent);
#endif
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::notifyMonitors() {
  ActiveOperation operation(*this);
  if (!operation) {
    return;
  }
 
#if THREADS
  LockGuard<Mutex> guard(m_Lock);
  notifyMonitorsLocked();
#endif
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::notifyMonitorsLocked() {
#if THREADS
  m_Monitors.notify();
#endif
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::writeLocked(const T* buffer, size_t count) {
  const bool wasReadable = m_bCanRead && m_DataSize;
  size_t countSoFar = 0;
  while (count) {
    size_t numberCopied = 0;
    if (count >= m_SegmentSize) {
      addSegment(buffer, m_SegmentSize);
      numberCopied = m_SegmentSize;
    } else if (!m_Segments.count()) {
      addSegment(buffer, count);
      numberCopied = count;
    } else {
      Segment* pSegment = m_Segments.popBack();
      if (pSegment->size == m_SegmentSize) {
        m_Segments.pushBack(pSegment);
        addSegment(buffer, count);
        numberCopied = count;
      } else {
        T* start = &pSegment->data[pSegment->size];
        size_t availableSpace = m_SegmentSize - pSegment->size;
        if (availableSpace > count) {
          availableSpace = count;
        }
        pedigree_std::copy(start, buffer, availableSpace);
        pSegment->size += availableSpace;
        m_Segments.pushBack(pSegment);
 
        if (availableSpace < count) {
          addSegment(&buffer[availableSpace], count - availableSpace);
        }
        numberCopied = count;
      }
    }
 
    countSoFar += numberCopied;
    m_DataSize += numberCopied;
    buffer += numberCopied;
    count -= numberCopied;
  }
 
  if (!wasReadable && m_bCanRead && m_DataSize) {
    m_ReadableGeneration += 1;
  }
 
  return countSoFar;
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::addSegment(const T* buffer, size_t count) {
  // Called with lock taken.
  Segment* pNewSegment = new Segment();
  pedigree_std::copy(pNewSegment->data, buffer, count);
  pNewSegment->size = count;
  m_Segments.pushBack(pNewSegment);
}
 
#if HOSTED && PEDIGREE_HOSTED_SMOKE_TESTS
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::acquireHostedOperationLock() {
  m_Lock.acquire();
}
 
template <class T, bool allowShortOperation>
void Buffer<T, allowShortOperation>::releaseHostedOperationLock() {
  m_Lock.release();
}
 
template <class T, bool allowShortOperation>
size_t Buffer<T, allowShortOperation>::getHostedActiveOperationCount() {
  LockGuard<Mutex> guard(m_Lock);
  return m_ActiveOperations;
}
#endif
 
template class Buffer<uint8_t, false>;
template class Buffer<uint8_t, true>;
template class Buffer<char, false>;
template class Buffer<char, true>;
template class Buffer<size_t, false>;
template class Buffer<size_t, true>;