mozilla
/
gecko-dev
зеркало из https://github.com/mozilla/gecko-dev.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность
gecko-dev/xpcom/io/nsPipe3.cpp

1370 строки
36 KiB
C++
Исходник Ответственный История

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
#include "mozilla/Attributes.h"
#include "mozilla/ReentrantMonitor.h"
#include "nsIPipe.h"
#include "nsIEventTarget.h"
#include "nsISeekableStream.h"
#include "nsIProgrammingLanguage.h"
#include "nsSegmentedBuffer.h"
#include "nsStreamUtils.h"
#include "nsCOMPtr.h"
#include "nsCRT.h"
#include "prlog.h"
#include "nsIClassInfoImpl.h"
#include "nsAlgorithm.h"
#include "nsMemory.h"
#include "nsIAsyncInputStream.h"
#include "nsIAsyncOutputStream.h"
using namespace mozilla;
#ifdef LOG
#undef LOG
#endif
#if defined(PR_LOGGING)
//
// set NSPR_LOG_MODULES=nsPipe:5
//
static PRLogModuleInfo*
GetPipeLog()
{
static PRLogModuleInfo* sLog;
if (!sLog) {
sLog = PR_NewLogModule("nsPipe");
}
return sLog;
}
#define LOG(args) PR_LOG(GetPipeLog(), PR_LOG_DEBUG, args)
#else
#define LOG(args)
#endif
#define DEFAULT_SEGMENT_SIZE 4096
#define DEFAULT_SEGMENT_COUNT 16
class nsPipe;
class nsPipeEvents;
class nsPipeInputStream;
class nsPipeOutputStream;
//-----------------------------------------------------------------------------
// this class is used to delay notifications until the end of a particular
// scope. it helps avoid the complexity of issuing callbacks while inside
// a critical section.
class nsPipeEvents
{
public:
nsPipeEvents() { }
~nsPipeEvents();
inline void NotifyInputReady(nsIAsyncInputStream* aStream,
nsIInputStreamCallback* aCallback)
{
NS_ASSERTION(!mInputCallback, "already have an input event");
mInputStream = aStream;
mInputCallback = aCallback;
}
inline void NotifyOutputReady(nsIAsyncOutputStream* aStream,
nsIOutputStreamCallback* aCallback)
{
NS_ASSERTION(!mOutputCallback, "already have an output event");
mOutputStream = aStream;
mOutputCallback = aCallback;
}
private:
nsCOMPtr<nsIAsyncInputStream> mInputStream;
nsCOMPtr<nsIInputStreamCallback> mInputCallback;
nsCOMPtr<nsIAsyncOutputStream> mOutputStream;
nsCOMPtr<nsIOutputStreamCallback> mOutputCallback;
};
//-----------------------------------------------------------------------------
// the input end of a pipe (allocated as a member of the pipe).
class nsPipeInputStream
: public nsIAsyncInputStream
, public nsISeekableStream
, public nsISearchableInputStream
, public nsIClassInfo
{
public:
// since this class will be allocated as a member of the pipe, we do not
// need our own ref count. instead, we share the lifetime (the ref count)
// of the entire pipe. this macro is just convenience since it does not
// declare a mRefCount variable; however, don't let the name fool you...
// we are not inheriting from nsPipe ;-)
NS_DECL_ISUPPORTS_INHERITED
NS_DECL_NSIINPUTSTREAM
NS_DECL_NSIASYNCINPUTSTREAM
NS_DECL_NSISEEKABLESTREAM
NS_DECL_NSISEARCHABLEINPUTSTREAM
NS_DECL_NSICLASSINFO
nsPipeInputStream(nsPipe* aPipe)
: mPipe(aPipe)
, mReaderRefCnt(0)
, mLogicalOffset(0)
, mBlocking(true)
, mBlocked(false)
, mAvailable(0)
, mCallbackFlags(0)
{ }
nsresult Fill();
void SetNonBlocking(bool aNonBlocking)
{
mBlocking = !aNonBlocking;
}
uint32_t Available()
{
return mAvailable;
}
void ReduceAvailable(uint32_t aAvail)
{
mAvailable -= aAvail;
}
// synchronously wait for the pipe to become readable.
nsresult Wait();
// these functions return true to indicate that the pipe's monitor should
// be notified, to wake up a blocked reader if any.
bool OnInputReadable(uint32_t aBytesWritten, nsPipeEvents&);
bool OnInputException(nsresult, nsPipeEvents&);
private:
nsPipe* mPipe;
// separate refcnt so that we know when to close the consumer
mozilla::ThreadSafeAutoRefCnt mReaderRefCnt;
int64_t mLogicalOffset;
bool mBlocking;
// these variables can only be accessed while inside the pipe's monitor
bool mBlocked;
uint32_t mAvailable;
nsCOMPtr<nsIInputStreamCallback> mCallback;
uint32_t mCallbackFlags;
};
//-----------------------------------------------------------------------------
// the output end of a pipe (allocated as a member of the pipe).
class nsPipeOutputStream
: public nsIAsyncOutputStream
, public nsIClassInfo
{
public:
// since this class will be allocated as a member of the pipe, we do not
// need our own ref count. instead, we share the lifetime (the ref count)
// of the entire pipe. this macro is just convenience since it does not
// declare a mRefCount variable; however, don't let the name fool you...
// we are not inheriting from nsPipe ;-)
NS_DECL_ISUPPORTS_INHERITED
NS_DECL_NSIOUTPUTSTREAM
NS_DECL_NSIASYNCOUTPUTSTREAM
NS_DECL_NSICLASSINFO
nsPipeOutputStream(nsPipe* aPipe)
: mPipe(aPipe)
, mWriterRefCnt(0)
, mLogicalOffset(0)
, mBlocking(true)
, mBlocked(false)
, mWritable(true)
, mCallbackFlags(0)
{ }
void SetNonBlocking(bool aNonBlocking)
{
mBlocking = !aNonBlocking;
}
void SetWritable(bool aWritable)
{
mWritable = aWritable;
}
// synchronously wait for the pipe to become writable.
nsresult Wait();
// these functions return true to indicate that the pipe's monitor should
// be notified, to wake up a blocked writer if any.
bool OnOutputWritable(nsPipeEvents&);
bool OnOutputException(nsresult, nsPipeEvents&);
private:
nsPipe* mPipe;
// separate refcnt so that we know when to close the producer
mozilla::ThreadSafeAutoRefCnt mWriterRefCnt;
int64_t mLogicalOffset;
bool mBlocking;
// these variables can only be accessed while inside the pipe's monitor
bool mBlocked;
bool mWritable;
nsCOMPtr<nsIOutputStreamCallback> mCallback;
uint32_t mCallbackFlags;
};
//-----------------------------------------------------------------------------
class nsPipe MOZ_FINAL : public nsIPipe
{
public:
friend class nsPipeInputStream;
friend class nsPipeOutputStream;
NS_DECL_THREADSAFE_ISUPPORTS
NS_DECL_NSIPIPE
// nsPipe methods:
nsPipe();
private:
~nsPipe();
public:
//
// methods below may only be called while inside the pipe's monitor
//
void PeekSegment(uint32_t aIndex, char*& aCursor, char*& aLimit);
//
// methods below may be called while outside the pipe's monitor
//
nsresult GetReadSegment(const char*& aSegment, uint32_t& aSegmentLen);
void AdvanceReadCursor(uint32_t aCount);
nsresult GetWriteSegment(char*& aSegment, uint32_t& aSegmentLen);
void AdvanceWriteCursor(uint32_t aCount);
void OnPipeException(nsresult aReason, bool aOutputOnly = false);
protected:
// We can't inherit from both nsIInputStream and nsIOutputStream
// because they collide on their Close method. Consequently we nest their
// implementations to avoid the extra object allocation.
nsPipeInputStream mInput;
nsPipeOutputStream mOutput;
ReentrantMonitor mReentrantMonitor;
nsSegmentedBuffer mBuffer;
char* mReadCursor;
char* mReadLimit;
int32_t mWriteSegment;
char* mWriteCursor;
char* mWriteLimit;
nsresult mStatus;
bool mInited;
};
//
// NOTES on buffer architecture:
//
// +-----------------+ - - mBuffer.GetSegment(0)
// | |
// + - - - - - - - - + - - mReadCursor
// |/////////////////|
// |/////////////////|
// |/////////////////|
// |/////////////////|
// +-----------------+ - - mReadLimit
// |
// +-----------------+
// |/////////////////|
// |/////////////////|
// |/////////////////|
// |/////////////////|
// |/////////////////|
// |/////////////////|
// +-----------------+
// |
// +-----------------+ - - mBuffer.GetSegment(mWriteSegment)
// |/////////////////|
// |/////////////////|
// |/////////////////|
// + - - - - - - - - + - - mWriteCursor
// | |
// | |
// +-----------------+ - - mWriteLimit
//
// (shaded region contains data)
//
// NOTE: on some systems (notably OS/2), the heap allocator uses an arena for
// small allocations (e.g., 64 byte allocations). this means that buffers may
// be allocated back-to-back. in the diagram above, for example, mReadLimit
// would actually be pointing at the beginning of the next segment. when
// making changes to this file, please keep this fact in mind.
//
//-----------------------------------------------------------------------------
// nsPipe methods:
//-----------------------------------------------------------------------------
nsPipe::nsPipe()
: mInput(MOZ_THIS_IN_INITIALIZER_LIST())
, mOutput(MOZ_THIS_IN_INITIALIZER_LIST())
, mReentrantMonitor("nsPipe.mReentrantMonitor")
, mReadCursor(nullptr)
, mReadLimit(nullptr)
, mWriteSegment(-1)
, mWriteCursor(nullptr)
, mWriteLimit(nullptr)
, mStatus(NS_OK)
, mInited(false)
{
}
nsPipe::~nsPipe()
{
}
NS_IMPL_ISUPPORTS(nsPipe, nsIPipe)
NS_IMETHODIMP
nsPipe::Init(bool aNonBlockingIn,
bool aNonBlockingOut,
uint32_t aSegmentSize,
uint32_t aSegmentCount)
{
mInited = true;
if (aSegmentSize == 0) {
aSegmentSize = DEFAULT_SEGMENT_SIZE;
}
if (aSegmentCount == 0) {
aSegmentCount = DEFAULT_SEGMENT_COUNT;
}
// protect against overflow
uint32_t maxCount = uint32_t(-1) / aSegmentSize;
if (aSegmentCount > maxCount) {
aSegmentCount = maxCount;
}
nsresult rv = mBuffer.Init(aSegmentSize, aSegmentSize * aSegmentCount);
if (NS_FAILED(rv)) {
return rv;
}
mInput.SetNonBlocking(aNonBlockingIn);
mOutput.SetNonBlocking(aNonBlockingOut);
return NS_OK;
}
NS_IMETHODIMP
nsPipe::GetInputStream(nsIAsyncInputStream** aInputStream)
{
NS_ADDREF(*aInputStream = &mInput);
return NS_OK;
}
NS_IMETHODIMP
nsPipe::GetOutputStream(nsIAsyncOutputStream** aOutputStream)
{
if (NS_WARN_IF(!mInited)) {
return NS_ERROR_NOT_INITIALIZED;
}
NS_ADDREF(*aOutputStream = &mOutput);
return NS_OK;
}
void
nsPipe::PeekSegment(uint32_t aIndex, char*& aCursor, char*& aLimit)
{
if (aIndex == 0) {
NS_ASSERTION(!mReadCursor || mBuffer.GetSegmentCount(), "unexpected state");
aCursor = mReadCursor;
aLimit = mReadLimit;
} else {
uint32_t numSegments = mBuffer.GetSegmentCount();
if (aIndex >= numSegments) {
aCursor = aLimit = nullptr;
} else {
aCursor = mBuffer.GetSegment(aIndex);
if (mWriteSegment == (int32_t)aIndex) {
aLimit = mWriteCursor;
} else {
aLimit = aCursor + mBuffer.GetSegmentSize();
}
}
}
}
nsresult
nsPipe::GetReadSegment(const char*& aSegment, uint32_t& aSegmentLen)
{
ReentrantMonitorAutoEnter mon(mReentrantMonitor);
if (mReadCursor == mReadLimit) {
return NS_FAILED(mStatus) ? mStatus : NS_BASE_STREAM_WOULD_BLOCK;
}
aSegment = mReadCursor;
aSegmentLen = mReadLimit - mReadCursor;
return NS_OK;
}
void
nsPipe::AdvanceReadCursor(uint32_t aBytesRead)
{
NS_ASSERTION(aBytesRead, "don't call if no bytes read");
nsPipeEvents events;
{
ReentrantMonitorAutoEnter mon(mReentrantMonitor);
LOG(("III advancing read cursor by %u\n", aBytesRead));
NS_ASSERTION(aBytesRead <= mBuffer.GetSegmentSize(), "read too much");
mReadCursor += aBytesRead;
NS_ASSERTION(mReadCursor <= mReadLimit, "read cursor exceeds limit");
mInput.ReduceAvailable(aBytesRead);
if (mReadCursor == mReadLimit) {
// we've reached the limit of how much we can read from this segment.
// if at the end of this segment, then we must discard this segment.
// if still writing in this segment then bail because we're not done
// with the segment and have to wait for now...
if (mWriteSegment == 0 && mWriteLimit > mWriteCursor) {
NS_ASSERTION(mReadLimit == mWriteCursor, "unexpected state");
return;
}
// shift write segment index (-1 indicates an empty buffer).
--mWriteSegment;
// done with this segment
mBuffer.DeleteFirstSegment();
LOG(("III deleting first segment\n"));
if (mWriteSegment == -1) {
// buffer is completely empty
mReadCursor = nullptr;
mReadLimit = nullptr;
mWriteCursor = nullptr;
mWriteLimit = nullptr;
} else {
// advance read cursor and limit to next buffer segment
mReadCursor = mBuffer.GetSegment(0);
if (mWriteSegment == 0) {
mReadLimit = mWriteCursor;
} else {
mReadLimit = mReadCursor + mBuffer.GetSegmentSize();
}
}
// we've free'd up a segment, so notify output stream that pipe has
// room for a new segment.
if (mOutput.OnOutputWritable(events)) {
mon.Notify();
}
}
}
}
nsresult
nsPipe::GetWriteSegment(char*& aSegment, uint32_t& aSegmentLen)
{
ReentrantMonitorAutoEnter mon(mReentrantMonitor);
if (NS_FAILED(mStatus)) {
return mStatus;
}
// write cursor and limit may both be null indicating an empty buffer.
if (mWriteCursor == mWriteLimit) {
char* seg = mBuffer.AppendNewSegment();
// pipe is full
if (!seg) {
return NS_BASE_STREAM_WOULD_BLOCK;
}
LOG(("OOO appended new segment\n"));
mWriteCursor = seg;
mWriteLimit = mWriteCursor + mBuffer.GetSegmentSize();
++mWriteSegment;
}
// make sure read cursor is initialized
if (!mReadCursor) {
NS_ASSERTION(mWriteSegment == 0, "unexpected null read cursor");
mReadCursor = mReadLimit = mWriteCursor;
}
// check to see if we can roll-back our read and write cursors to the
// beginning of the current/first segment. this is purely an optimization.
if (mReadCursor == mWriteCursor && mWriteSegment == 0) {
char* head = mBuffer.GetSegment(0);
LOG(("OOO rolling back write cursor %u bytes\n", mWriteCursor - head));
mWriteCursor = mReadCursor = mReadLimit = head;
}
aSegment = mWriteCursor;
aSegmentLen = mWriteLimit - mWriteCursor;
return NS_OK;
}
void
nsPipe::AdvanceWriteCursor(uint32_t aBytesWritten)
{
NS_ASSERTION(aBytesWritten, "don't call if no bytes written");
nsPipeEvents events;
{
ReentrantMonitorAutoEnter mon(mReentrantMonitor);
LOG(("OOO advancing write cursor by %u\n", aBytesWritten));
char* newWriteCursor = mWriteCursor + aBytesWritten;
NS_ASSERTION(newWriteCursor <= mWriteLimit, "write cursor exceeds limit");
// update read limit if reading in the same segment
if (mWriteSegment == 0 && mReadLimit == mWriteCursor) {
mReadLimit = newWriteCursor;
}
mWriteCursor = newWriteCursor;
// The only way mReadCursor == mWriteCursor is if:
//
// - mReadCursor is at the start of a segment (which, based on how
// nsSegmentedBuffer works, means that this segment is the "first"
// segment)
// - mWriteCursor points at the location past the end of the current
// write segment (so the current write filled the current write
// segment, so we've incremented mWriteCursor to point past the end
// of it)
// - the segment to which data has just been written is located
// exactly one segment's worth of bytes before the first segment
// where mReadCursor is located
//
// Consequently, the byte immediately after the end of the current
// write segment is the first byte of the first segment, so
// mReadCursor == mWriteCursor. (Another way to think about this is
// to consider the buffer architecture diagram above, but consider it
// with an arena allocator which allocates from the *end* of the
// arena to the *beginning* of the arena.)
NS_ASSERTION(mReadCursor != mWriteCursor ||
(mBuffer.GetSegment(0) == mReadCursor &&
mWriteCursor == mWriteLimit),
"read cursor is bad");
// update the writable flag on the output stream
if (mWriteCursor == mWriteLimit) {
if (mBuffer.GetSize() >= mBuffer.GetMaxSize()) {
mOutput.SetWritable(false);
}
}
// notify input stream that pipe now contains additional data
if (mInput.OnInputReadable(aBytesWritten, events)) {
mon.Notify();
}
}
}
void
nsPipe::OnPipeException(nsresult aReason, bool aOutputOnly)
{
LOG(("PPP nsPipe::OnPipeException [reason=%x output-only=%d]\n",
aReason, aOutputOnly));
nsPipeEvents events;
{
ReentrantMonitorAutoEnter mon(mReentrantMonitor);
// if we've already hit an exception, then ignore this one.
if (NS_FAILED(mStatus)) {
return;
}
mStatus = aReason;
// an output-only exception applies to the input end if the pipe has
// zero bytes available.
if (aOutputOnly && !mInput.Available()) {
aOutputOnly = false;
}
if (!aOutputOnly)
if (mInput.OnInputException(aReason, events)) {
mon.Notify();
}
if (mOutput.OnOutputException(aReason, events)) {
mon.Notify();
}
}
}
//-----------------------------------------------------------------------------
// nsPipeEvents methods:
//-----------------------------------------------------------------------------
nsPipeEvents::~nsPipeEvents()
{
// dispatch any pending events
if (mInputCallback) {
mInputCallback->OnInputStreamReady(mInputStream);
mInputCallback = 0;
mInputStream = 0;
}
if (mOutputCallback) {
mOutputCallback->OnOutputStreamReady(mOutputStream);
mOutputCallback = 0;
mOutputStream = 0;
}
}
//-----------------------------------------------------------------------------
// nsPipeInputStream methods:
//-----------------------------------------------------------------------------
NS_IMPL_QUERY_INTERFACE(nsPipeInputStream,
nsIInputStream,
nsIAsyncInputStream,
nsISeekableStream,
nsISearchableInputStream,
nsIClassInfo)
NS_IMPL_CI_INTERFACE_GETTER(nsPipeInputStream,
nsIInputStream,
nsIAsyncInputStream,
nsISeekableStream,
nsISearchableInputStream)
NS_IMPL_THREADSAFE_CI(nsPipeInputStream)
nsresult
nsPipeInputStream::Wait()
{
NS_ASSERTION(mBlocking, "wait on non-blocking pipe input stream");
ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
while (NS_SUCCEEDED(mPipe->mStatus) && (mAvailable == 0)) {
LOG(("III pipe input: waiting for data\n"));
mBlocked = true;
mon.Wait();
mBlocked = false;
LOG(("III pipe input: woke up [pipe-status=%x available=%u]\n",
mPipe->mStatus, mAvailable));
}
return mPipe->mStatus == NS_BASE_STREAM_CLOSED ? NS_OK : mPipe->mStatus;
}
bool
nsPipeInputStream::OnInputReadable(uint32_t aBytesWritten, nsPipeEvents& aEvents)
{
bool result = false;
mAvailable += aBytesWritten;
if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) {
aEvents.NotifyInputReady(this, mCallback);
mCallback = 0;
mCallbackFlags = 0;
} else if (mBlocked) {
result = true;
}
return result;
}
bool
nsPipeInputStream::OnInputException(nsresult aReason, nsPipeEvents& aEvents)
{
LOG(("nsPipeInputStream::OnInputException [this=%x reason=%x]\n",
this, aReason));
bool result = false;
NS_ASSERTION(NS_FAILED(aReason), "huh? successful exception");
// force count of available bytes to zero.
mAvailable = 0;
if (mCallback) {
aEvents.NotifyInputReady(this, mCallback);
mCallback = 0;
mCallbackFlags = 0;
} else if (mBlocked) {
result = true;
}
return result;
}
NS_IMETHODIMP_(MozExternalRefCountType)
nsPipeInputStream::AddRef(void)
{
++mReaderRefCnt;
return mPipe->AddRef();
}
NS_IMETHODIMP_(MozExternalRefCountType)
nsPipeInputStream::Release(void)
{
if (--mReaderRefCnt == 0) {
Close();
}
return mPipe->Release();
}
NS_IMETHODIMP
nsPipeInputStream::CloseWithStatus(nsresult aReason)
{
LOG(("III CloseWithStatus [this=%x reason=%x]\n", this, aReason));
if (NS_SUCCEEDED(aReason)) {
aReason = NS_BASE_STREAM_CLOSED;
}
mPipe->OnPipeException(aReason);
return NS_OK;
}
NS_IMETHODIMP
nsPipeInputStream::Close()
{
return CloseWithStatus(NS_BASE_STREAM_CLOSED);
}
NS_IMETHODIMP
nsPipeInputStream::Available(uint64_t* aResult)
{
// nsPipeInputStream supports under 4GB stream only
ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
// return error if pipe closed
if (!mAvailable && NS_FAILED(mPipe->mStatus)) {
return mPipe->mStatus;
}
*aResult = (uint64_t)mAvailable;
return NS_OK;
}
NS_IMETHODIMP
nsPipeInputStream::ReadSegments(nsWriteSegmentFun aWriter,
void* aClosure,
uint32_t aCount,
uint32_t* aReadCount)
{
LOG(("III ReadSegments [this=%x count=%u]\n", this, aCount));
nsresult rv = NS_OK;
const char* segment;
uint32_t segmentLen;
*aReadCount = 0;
while (aCount) {
rv = mPipe->GetReadSegment(segment, segmentLen);
if (NS_FAILED(rv)) {
// ignore this error if we've already read something.
if (*aReadCount > 0) {
rv = NS_OK;
break;
}
if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
// pipe is empty
if (!mBlocking) {
break;
}
// wait for some data to be written to the pipe
rv = Wait();
if (NS_SUCCEEDED(rv)) {
continue;
}
}
// ignore this error, just return.
if (rv == NS_BASE_STREAM_CLOSED) {
rv = NS_OK;
break;
}
mPipe->OnPipeException(rv);
break;
}
// read no more than aCount
if (segmentLen > aCount) {
segmentLen = aCount;
}
uint32_t writeCount, originalLen = segmentLen;
while (segmentLen) {
writeCount = 0;
rv = aWriter(this, aClosure, segment, *aReadCount, segmentLen, &writeCount);
if (NS_FAILED(rv) || writeCount == 0) {
aCount = 0;
// any errors returned from the writer end here: do not
// propagate to the caller of ReadSegments.
rv = NS_OK;
break;
}
NS_ASSERTION(writeCount <= segmentLen, "wrote more than expected");
segment += writeCount;
segmentLen -= writeCount;
aCount -= writeCount;
*aReadCount += writeCount;
mLogicalOffset += writeCount;
}
if (segmentLen < originalLen) {
mPipe->AdvanceReadCursor(originalLen - segmentLen);
}
}
return rv;
}
NS_IMETHODIMP
nsPipeInputStream::Read(char* aToBuf, uint32_t aBufLen, uint32_t* aReadCount)
{
return ReadSegments(NS_CopySegmentToBuffer, aToBuf, aBufLen, aReadCount);
}
NS_IMETHODIMP
nsPipeInputStream::IsNonBlocking(bool* aNonBlocking)
{
*aNonBlocking = !mBlocking;
return NS_OK;
}
NS_IMETHODIMP
nsPipeInputStream::AsyncWait(nsIInputStreamCallback* aCallback,
uint32_t aFlags,
uint32_t aRequestedCount,
nsIEventTarget* aTarget)
{
LOG(("III AsyncWait [this=%x]\n", this));
nsPipeEvents pipeEvents;
{
ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
// replace a pending callback
mCallback = 0;
mCallbackFlags = 0;
if (!aCallback) {
return NS_OK;
}
nsCOMPtr<nsIInputStreamCallback> proxy;
if (aTarget) {
proxy = NS_NewInputStreamReadyEvent(aCallback, aTarget);
aCallback = proxy;
}
if (NS_FAILED(mPipe->mStatus) ||
(mAvailable && !(aFlags & WAIT_CLOSURE_ONLY))) {
// stream is already closed or readable; post event.
pipeEvents.NotifyInputReady(this, aCallback);
} else {
// queue up callback object to be notified when data becomes available
mCallback = aCallback;
mCallbackFlags = aFlags;
}
}
return NS_OK;
}
NS_IMETHODIMP
nsPipeInputStream::Seek(int32_t aWhence, int64_t aOffset)
{
NS_NOTREACHED("nsPipeInputStream::Seek");
return NS_ERROR_NOT_IMPLEMENTED;
}
NS_IMETHODIMP
nsPipeInputStream::Tell(int64_t* aOffset)
{
ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
// return error if pipe closed
if (!mAvailable && NS_FAILED(mPipe->mStatus)) {
return mPipe->mStatus;
}
*aOffset = mLogicalOffset;
return NS_OK;
}
NS_IMETHODIMP
nsPipeInputStream::SetEOF()
{
NS_NOTREACHED("nsPipeInputStream::SetEOF");
return NS_ERROR_NOT_IMPLEMENTED;
}
#define COMPARE(s1, s2, i) \
(aIgnoreCase \
? nsCRT::strncasecmp((const char *)s1, (const char *)s2, (uint32_t)i) \
: nsCRT::strncmp((const char *)s1, (const char *)s2, (uint32_t)i))
NS_IMETHODIMP
nsPipeInputStream::Search(const char* aForString,
bool aIgnoreCase,
bool* aFound,
uint32_t* aOffsetSearchedTo)
{
LOG(("III Search [for=%s ic=%u]\n", aForString, aIgnoreCase));
ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
char* cursor1;
char* limit1;
uint32_t index = 0, offset = 0;
uint32_t strLen = strlen(aForString);
mPipe->PeekSegment(0, cursor1, limit1);
if (cursor1 == limit1) {
*aFound = false;
*aOffsetSearchedTo = 0;
LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
return NS_OK;
}
while (true) {
uint32_t i, len1 = limit1 - cursor1;
// check if the string is in the buffer segment
for (i = 0; i < len1 - strLen + 1; i++) {
if (COMPARE(&cursor1[i], aForString, strLen) == 0) {
*aFound = true;
*aOffsetSearchedTo = offset + i;
LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
return NS_OK;
}
}
// get the next segment
char* cursor2;
char* limit2;
uint32_t len2;
index++;
offset += len1;
mPipe->PeekSegment(index, cursor2, limit2);
if (cursor2 == limit2) {
*aFound = false;
*aOffsetSearchedTo = offset - strLen + 1;
LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
return NS_OK;
}
len2 = limit2 - cursor2;
// check if the string is straddling the next buffer segment
uint32_t lim = XPCOM_MIN(strLen, len2 + 1);
for (i = 0; i < lim; ++i) {
uint32_t strPart1Len = strLen - i - 1;
uint32_t strPart2Len = strLen - strPart1Len;
const char* strPart2 = &aForString[strLen - strPart2Len];
uint32_t bufSeg1Offset = len1 - strPart1Len;
if (COMPARE(&cursor1[bufSeg1Offset], aForString, strPart1Len) == 0 &&
COMPARE(cursor2, strPart2, strPart2Len) == 0) {
*aFound = true;
*aOffsetSearchedTo = offset - strPart1Len;
LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
return NS_OK;
}
}
// finally continue with the next buffer
cursor1 = cursor2;
limit1 = limit2;
}
NS_NOTREACHED("can't get here");
return NS_ERROR_UNEXPECTED; // keep compiler happy
}
//-----------------------------------------------------------------------------
// nsPipeOutputStream methods:
//-----------------------------------------------------------------------------
NS_IMPL_QUERY_INTERFACE(nsPipeOutputStream,
nsIOutputStream,
nsIAsyncOutputStream,
nsIClassInfo)
NS_IMPL_CI_INTERFACE_GETTER(nsPipeOutputStream,
nsIOutputStream,
nsIAsyncOutputStream)
NS_IMPL_THREADSAFE_CI(nsPipeOutputStream)
nsresult
nsPipeOutputStream::Wait()
{
NS_ASSERTION(mBlocking, "wait on non-blocking pipe output stream");
ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
if (NS_SUCCEEDED(mPipe->mStatus) && !mWritable) {
LOG(("OOO pipe output: waiting for space\n"));
mBlocked = true;
mon.Wait();
mBlocked = false;
LOG(("OOO pipe output: woke up [pipe-status=%x writable=%u]\n",
mPipe->mStatus, mWritable));
}
return mPipe->mStatus == NS_BASE_STREAM_CLOSED ? NS_OK : mPipe->mStatus;
}
bool
nsPipeOutputStream::OnOutputWritable(nsPipeEvents& aEvents)
{
bool result = false;
mWritable = true;
if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) {
aEvents.NotifyOutputReady(this, mCallback);
mCallback = 0;
mCallbackFlags = 0;
} else if (mBlocked) {
result = true;
}
return result;
}
bool
nsPipeOutputStream::OnOutputException(nsresult aReason, nsPipeEvents& aEvents)
{
LOG(("nsPipeOutputStream::OnOutputException [this=%x reason=%x]\n",
this, aReason));
bool result = false;
NS_ASSERTION(NS_FAILED(aReason), "huh? successful exception");
mWritable = false;
if (mCallback) {
aEvents.NotifyOutputReady(this, mCallback);
mCallback = 0;
mCallbackFlags = 0;
} else if (mBlocked) {
result = true;
}
return result;
}
NS_IMETHODIMP_(MozExternalRefCountType)
nsPipeOutputStream::AddRef()
{
++mWriterRefCnt;
return mPipe->AddRef();
}
NS_IMETHODIMP_(MozExternalRefCountType)
nsPipeOutputStream::Release()
{
if (--mWriterRefCnt == 0) {
Close();
}
return mPipe->Release();
}
NS_IMETHODIMP
nsPipeOutputStream::CloseWithStatus(nsresult aReason)
{
LOG(("OOO CloseWithStatus [this=%x reason=%x]\n", this, aReason));
if (NS_SUCCEEDED(aReason)) {
aReason = NS_BASE_STREAM_CLOSED;
}
// input stream may remain open
mPipe->OnPipeException(aReason, true);
return NS_OK;
}
NS_IMETHODIMP
nsPipeOutputStream::Close()
{
return CloseWithStatus(NS_BASE_STREAM_CLOSED);
}
NS_IMETHODIMP
nsPipeOutputStream::WriteSegments(nsReadSegmentFun aReader,
void* aClosure,
uint32_t aCount,
uint32_t* aWriteCount)
{
LOG(("OOO WriteSegments [this=%x count=%u]\n", this, aCount));
nsresult rv = NS_OK;
char* segment;
uint32_t segmentLen;
*aWriteCount = 0;
while (aCount) {
rv = mPipe->GetWriteSegment(segment, segmentLen);
if (NS_FAILED(rv)) {
if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
// pipe is full
if (!mBlocking) {
// ignore this error if we've already written something
if (*aWriteCount > 0) {
rv = NS_OK;
}
break;
}
// wait for the pipe to have an empty segment.
rv = Wait();
if (NS_SUCCEEDED(rv)) {
continue;
}
}
mPipe->OnPipeException(rv);
break;
}
// write no more than aCount
if (segmentLen > aCount) {
segmentLen = aCount;
}
uint32_t readCount, originalLen = segmentLen;
while (segmentLen) {
readCount = 0;
rv = aReader(this, aClosure, segment, *aWriteCount, segmentLen, &readCount);
if (NS_FAILED(rv) || readCount == 0) {
aCount = 0;
// any errors returned from the aReader end here: do not
// propagate to the caller of WriteSegments.
rv = NS_OK;
break;
}
NS_ASSERTION(readCount <= segmentLen, "read more than expected");
segment += readCount;
segmentLen -= readCount;
aCount -= readCount;
*aWriteCount += readCount;
mLogicalOffset += readCount;
}
if (segmentLen < originalLen) {
mPipe->AdvanceWriteCursor(originalLen - segmentLen);
}
}
return rv;
}
static NS_METHOD
nsReadFromRawBuffer(nsIOutputStream* aOutStr,
void* aClosure,
char* aToRawSegment,
uint32_t aOffset,
uint32_t aCount,
uint32_t* aReadCount)
{
const char* fromBuf = (const char*)aClosure;
memcpy(aToRawSegment, &fromBuf[aOffset], aCount);
*aReadCount = aCount;
return NS_OK;
}
NS_IMETHODIMP
nsPipeOutputStream::Write(const char* aFromBuf,
uint32_t aBufLen,
uint32_t* aWriteCount)
{
return WriteSegments(nsReadFromRawBuffer, (void*)aFromBuf, aBufLen, aWriteCount);
}
NS_IMETHODIMP
nsPipeOutputStream::Flush(void)
{
// nothing to do
return NS_OK;
}
static NS_METHOD
nsReadFromInputStream(nsIOutputStream* aOutStr,
void* aClosure,
char* aToRawSegment,
uint32_t aOffset,
uint32_t aCount,
uint32_t* aReadCount)
{
nsIInputStream* fromStream = (nsIInputStream*)aClosure;
return fromStream->Read(aToRawSegment, aCount, aReadCount);
}
NS_IMETHODIMP
nsPipeOutputStream::WriteFrom(nsIInputStream* aFromStream,
uint32_t aCount,
uint32_t* aWriteCount)
{
return WriteSegments(nsReadFromInputStream, aFromStream, aCount, aWriteCount);
}
NS_IMETHODIMP
nsPipeOutputStream::IsNonBlocking(bool* aNonBlocking)
{
*aNonBlocking = !mBlocking;
return NS_OK;
}
NS_IMETHODIMP
nsPipeOutputStream::AsyncWait(nsIOutputStreamCallback* aCallback,
uint32_t aFlags,
uint32_t aRequestedCount,
nsIEventTarget* aTarget)
{
LOG(("OOO AsyncWait [this=%x]\n", this));
nsPipeEvents pipeEvents;
{
ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
// replace a pending callback
mCallback = 0;
mCallbackFlags = 0;
if (!aCallback) {
return NS_OK;
}
nsCOMPtr<nsIOutputStreamCallback> proxy;
if (aTarget) {
proxy = NS_NewOutputStreamReadyEvent(aCallback, aTarget);
aCallback = proxy;
}
if (NS_FAILED(mPipe->mStatus) ||
(mWritable && !(aFlags & WAIT_CLOSURE_ONLY))) {
// stream is already closed or writable; post event.
pipeEvents.NotifyOutputReady(this, aCallback);
} else {
// queue up callback object to be notified when data becomes available
mCallback = aCallback;
mCallbackFlags = aFlags;
}
}
return NS_OK;
}
////////////////////////////////////////////////////////////////////////////////
nsresult
NS_NewPipe(nsIInputStream** aPipeIn,
nsIOutputStream** aPipeOut,
uint32_t aSegmentSize,
uint32_t aMaxSize,
bool aNonBlockingInput,
bool aNonBlockingOutput)
{
if (aSegmentSize == 0) {
aSegmentSize = DEFAULT_SEGMENT_SIZE;
}
// Handle aMaxSize of UINT32_MAX as a special case
uint32_t segmentCount;
if (aMaxSize == UINT32_MAX) {
segmentCount = UINT32_MAX;
} else {
segmentCount = aMaxSize / aSegmentSize;
}
nsIAsyncInputStream* in;
nsIAsyncOutputStream* out;
nsresult rv = NS_NewPipe2(&in, &out, aNonBlockingInput, aNonBlockingOutput,
aSegmentSize, segmentCount);
if (NS_FAILED(rv)) {
return rv;
}
*aPipeIn = in;
*aPipeOut = out;
return NS_OK;
}
nsresult
NS_NewPipe2(nsIAsyncInputStream** aPipeIn,
nsIAsyncOutputStream** aPipeOut,
bool aNonBlockingInput,
bool aNonBlockingOutput,
uint32_t aSegmentSize,
uint32_t aSegmentCount)
{
nsresult rv;
nsPipe* pipe = new nsPipe();
if (!pipe) {
return NS_ERROR_OUT_OF_MEMORY;
}
rv = pipe->Init(aNonBlockingInput,
aNonBlockingOutput,
aSegmentSize,
aSegmentCount);
if (NS_FAILED(rv)) {
NS_ADDREF(pipe);
NS_RELEASE(pipe);
return rv;
}
pipe->GetInputStream(aPipeIn);
pipe->GetOutputStream(aPipeOut);
return NS_OK;
}
nsresult
nsPipeConstructor(nsISupports* aOuter, REFNSIID aIID, void** aResult)
{
if (aOuter) {
return NS_ERROR_NO_AGGREGATION;
}
nsPipe* pipe = new nsPipe();
if (!pipe) {
return NS_ERROR_OUT_OF_MEMORY;
}
NS_ADDREF(pipe);
nsresult rv = pipe->QueryInterface(aIID, aResult);
NS_RELEASE(pipe);
return rv;
}
////////////////////////////////////////////////////////////////////////////////
Ссылка в новой задаче Показать git blame Копировать постоянную ссылку