Switch SkPDFStream's internal storage from SkStream to SkData

Motivation: This makes SkPDFStream thread-safe for two threads
serializing it at once, since a SkStream has an internal position.

Updated SkPDFFont, SkPDFGraphicState, and SkPDFPage's use of
SkPDFStream to use the SkData constructor rather than the SkStream
constructor (saving a memcpy).

BUG=skia:2683

Committed: https://skia.googlesource.com/skia/+/c1dfa14b645ae274780f026dd86c9b633fbdad06

R=mtklein@google.com, djsollen@google.com, rmistry@google.com, robertphillips@google.com

Author: halcanary@google.com

Review URL: https://codereview.chromium.org/340783013

14 files changed, 161 insertions, 149 deletions

diff --git a/src/core/SkStream.cpp b/src/core/SkStream.cpp
index ebaac9ab15..1022d183d9 100644
--- a/src/core/SkStream.cpp
+++ b/src/core/SkStream.cpp
@@ -8,10 +8,12 @@
 
 
 #include "SkStream.h"
+#include "SkStreamPriv.h"
 #include "SkData.h"
 #include "SkFixed.h"
 #include "SkString.h"
 #include "SkOSFile.h"
+#include "SkTypes.h"
 
 ///////////////////////////////////////////////////////////////////////////////
 
@@ -851,3 +853,57 @@ SkStreamAsset* SkStream::NewFromFile(const char path[]) {
     }
     return stream;
 }
+
+// Declared in SkStreamPriv.h:
+size_t SkCopyStreamToStorage(SkAutoMalloc* storage, SkStream* stream) {
+    SkASSERT(storage != NULL);
+    SkASSERT(stream != NULL);
+
+    if (stream->hasLength()) {
+        const size_t length = stream->getLength();
+        void* dst = storage->reset(length);
+        if (stream->read(dst, length) != length) {
+            return 0;
+        }
+        return length;
+    }
+
+    SkDynamicMemoryWStream tempStream;
+    // Arbitrary buffer size.
+    const size_t bufferSize = 256 * 1024; // 256KB
+    char buffer[bufferSize];
+    SkDEBUGCODE(size_t debugLength = 0;)
+    do {
+        size_t bytesRead = stream->read(buffer, bufferSize);
+        tempStream.write(buffer, bytesRead);
+        SkDEBUGCODE(debugLength += bytesRead);
+        SkASSERT(tempStream.bytesWritten() == debugLength);
+    } while (!stream->isAtEnd());
+    const size_t length = tempStream.bytesWritten();
+    void* dst = storage->reset(length);
+    tempStream.copyTo(dst);
+    return length;
+}
+
+// Declared in SkStreamPriv.h:
+SkData* SkCopyStreamToData(SkStream* stream) {
+    SkASSERT(stream != NULL);
+
+    if (stream->hasLength()) {
+        const size_t length = stream->getLength();
+        SkAutoMalloc dst(length);
+        if (stream->read(dst.get(), length) != length) {
+            return NULL;
+        }
+        return SkData::NewFromMalloc(dst.detach(), length);
+    }
+
+    SkDynamicMemoryWStream tempStream;
+    const size_t bufferSize = 4096;
+    char buffer[bufferSize];
+    do {
+        size_t bytesRead = stream->read(buffer, bufferSize);
+        tempStream.write(buffer, bytesRead);
+    } while (!stream->isAtEnd());
+    return tempStream.copyToData();
+}
diff --git a/src/images/SkStreamHelpers.h b/src/core/SkStreamPriv.h
index 008dd8e17a..5b5a73adef 100644
--- a/src/images/SkStreamHelpers.h
+++ b/src/core/SkStreamPriv.h
@@ -5,8 +5,8 @@
  * found in the LICENSE file.
  */
 
-#ifndef SkStreamHelpers_DEFINED
-#define SkStreamHelpers_DEFINED
+#ifndef SkStreamPriv_DEFINED
+#define SkStreamPriv_DEFINED
 
 class SkAutoMalloc;
 class SkStream;
@@ -23,14 +23,15 @@ class SkData;
  *  @return size_t Total number of bytes in the SkStream, which is also the
  *      number of bytes pointed to by storage->get(). Returns 0 on failure.
  */
-size_t CopyStreamToStorage(SkAutoMalloc* storage, SkStream* stream);
+size_t SkCopyStreamToStorage(SkAutoMalloc* storage, SkStream* stream);
 
 /**
- *  Copy the provided stream to an SkData variable. Used by SkImageDecoder_libktx.
+ *  Copy the provided stream to an SkData variable.
  *  @param stream SkStream to be copied into data.
- *  @return SkData* The resulting SkData after the copy. This data will have a
- *      ref count of one upon return and belongs to the caller. Returns NULL on failure.
+ *  @return SkData* The resulting SkData after the copy. This data
+ *      will have a ref count of one upon return and belongs to the
+ *      caller. Returns NULL on failure.
  */
-SkData *CopyStreamToData(SkStream* stream);
+SkData *SkCopyStreamToData(SkStream* stream);
 
-#endif // SkStreamHelpers_DEFINED
+#endif  // SkStreamPriv_DEFINED
diff --git a/src/images/SkImageDecoder_ktx.cpp b/src/images/SkImageDecoder_ktx.cpp
index effc1edcc1..c2606321c2 100644
--- a/src/images/SkImageDecoder_ktx.cpp
+++ b/src/images/SkImageDecoder_ktx.cpp
@@ -10,7 +10,7 @@
 #include "SkPixelRef.h"
 #include "SkScaledBitmapSampler.h"
 #include "SkStream.h"
-#include "SkStreamHelpers.h"
+#include "SkStreamPriv.h"
 #include "SkTypes.h"
 
 #include "ktx.h"
@@ -49,7 +49,7 @@ private:
 
 bool SkKTXImageDecoder::onDecode(SkStream* stream, SkBitmap* bm, Mode mode) {
     // TODO: Implement SkStream::copyToData() that's cheap for memory and file streams
-    SkAutoDataUnref data(CopyStreamToData(stream));
+    SkAutoDataUnref data(SkCopyStreamToData(stream));
     if (NULL == data) {
         return false;
     }
diff --git a/src/images/SkImageDecoder_libbmp.cpp b/src/images/SkImageDecoder_libbmp.cpp
index f9dd2472a0..7b87e40039 100644
--- a/src/images/SkImageDecoder_libbmp.cpp
+++ b/src/images/SkImageDecoder_libbmp.cpp
@@ -12,7 +12,7 @@
 #include "SkImageDecoder.h"
 #include "SkScaledBitmapSampler.h"
 #include "SkStream.h"
-#include "SkStreamHelpers.h"
+#include "SkStreamPriv.h"
 #include "SkTDArray.h"
 
 class SkBMPImageDecoder : public SkImageDecoder {
@@ -99,7 +99,7 @@ bool SkBMPImageDecoder::onDecode(SkStream* stream, SkBitmap* bm, Mode mode) {
     // Allocated space used to hold the data.
     SkAutoMalloc storage;
     // Byte length of all of the data.
-    const size_t length = CopyStreamToStorage(&storage, stream);
+    const size_t length = SkCopyStreamToStorage(&storage, stream);
     if (0 == length) {
         return 0;
     }
diff --git a/src/images/SkImageDecoder_libico.cpp b/src/images/SkImageDecoder_libico.cpp
index e6ae16f0bd..d415d2be6d 100644
--- a/src/images/SkImageDecoder_libico.cpp
+++ b/src/images/SkImageDecoder_libico.cpp
@@ -8,7 +8,7 @@
 #include "SkColorPriv.h"
 #include "SkImageDecoder.h"
 #include "SkStream.h"
-#include "SkStreamHelpers.h"
+#include "SkStreamPriv.h"
 #include "SkTypes.h"
 
 class SkICOImageDecoder : public SkImageDecoder {
@@ -75,7 +75,7 @@ static int calculateRowBytesFor8888(int w, int bitCount)
 bool SkICOImageDecoder::onDecode(SkStream* stream, SkBitmap* bm, Mode mode)
 {
     SkAutoMalloc autoMal;
-    const size_t length = CopyStreamToStorage(&autoMal, stream);
+    const size_t length = SkCopyStreamToStorage(&autoMal, stream);
     if (0 == length) {
         return false;
     }
diff --git a/src/images/SkImageDecoder_pkm.cpp b/src/images/SkImageDecoder_pkm.cpp
index d555c6afcc..738d98ec29 100644
--- a/src/images/SkImageDecoder_pkm.cpp
+++ b/src/images/SkImageDecoder_pkm.cpp
@@ -9,7 +9,7 @@
 #include "SkImageDecoder.h"
 #include "SkScaledBitmapSampler.h"
 #include "SkStream.h"
-#include "SkStreamHelpers.h"
+#include "SkStreamPriv.h"
 #include "SkTypes.h"
 
 #include "etc1.h"
@@ -33,7 +33,7 @@ private:
 
 bool SkPKMImageDecoder::onDecode(SkStream* stream, SkBitmap* bm, Mode mode) {
     SkAutoMalloc autoMal;
-    const size_t length = CopyStreamToStorage(&autoMal, stream);
+    const size_t length = SkCopyStreamToStorage(&autoMal, stream);
     if (0 == length) {
         return false;
     }
diff --git a/src/images/SkStreamHelpers.cpp b/src/images/SkStreamHelpers.cpp
deleted file mode 100644
index c7c66b4b0f..0000000000
--- a/src/images/SkStreamHelpers.cpp
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Copyright 2013 Google Inc.
- *
- * Use of this source code is governed by a BSD-style license that can be
- * found in the LICENSE file.
- */
-
-#include "SkData.h"
-#include "SkStream.h"
-#include "SkStreamHelpers.h"
-#include "SkTypes.h"
-
-size_t CopyStreamToStorage(SkAutoMalloc* storage, SkStream* stream) {
-    SkASSERT(storage != NULL);
-    SkASSERT(stream != NULL);
-
-    if (stream->hasLength()) {
-        const size_t length = stream->getLength();
-        void* dst = storage->reset(length);
-        if (stream->read(dst, length) != length) {
-            return 0;
-        }
-        return length;
-    }
-
-    SkDynamicMemoryWStream tempStream;
-    // Arbitrary buffer size.
-    const size_t bufferSize = 256 * 1024; // 256KB
-    char buffer[bufferSize];
-    SkDEBUGCODE(size_t debugLength = 0;)
-    do {
-        size_t bytesRead = stream->read(buffer, bufferSize);
-        tempStream.write(buffer, bytesRead);
-        SkDEBUGCODE(debugLength += bytesRead);
-        SkASSERT(tempStream.bytesWritten() == debugLength);
-    } while (!stream->isAtEnd());
-    const size_t length = tempStream.bytesWritten();
-    void* dst = storage->reset(length);
-    tempStream.copyTo(dst);
-    return length;
-}
-
-SkData *CopyStreamToData(SkStream* stream) {
-    SkASSERT(stream != NULL);
-
-    if (stream->hasLength()) {
-        const size_t length = stream->getLength();
-        void* dst = sk_malloc_throw(length);
-        if (stream->read(dst, length) != length) {
-            return 0;
-        }
-        return SkData::NewFromMalloc(dst, length);
-    }
-
-    SkDynamicMemoryWStream tempStream;
-    // Arbitrary buffer size.
-    const size_t bufferSize = 256 * 1024; // 256KB
-    char buffer[bufferSize];
-    SkDEBUGCODE(size_t debugLength = 0;)
-    do {
-        size_t bytesRead = stream->read(buffer, bufferSize);
-        tempStream.write(buffer, bytesRead);
-        SkDEBUGCODE(debugLength += bytesRead);
-        SkASSERT(tempStream.bytesWritten() == debugLength);
-    } while (!stream->isAtEnd());
-    return tempStream.copyToData();
-}
diff --git a/src/pdf/SkPDFFont.cpp b/src/pdf/SkPDFFont.cpp
index 014b328030..aa4b4b5534 100644
--- a/src/pdf/SkPDFFont.cpp
+++ b/src/pdf/SkPDFFont.cpp
@@ -150,8 +150,8 @@ int8_t hexToBin(uint8_t c) {
     return -1;
 }
 
-SkStream* handleType1Stream(SkStream* srcStream, size_t* headerLen,
-                            size_t* dataLen, size_t* trailerLen) {
+static SkData* handle_type1_stream(SkStream* srcStream, size_t* headerLen,
+                                   size_t* dataLen, size_t* trailerLen) {
     // srcStream may be backed by a file or a unseekable fd, so we may not be
     // able to use skip(), rewind(), or getMemoryBase().  read()ing through
     // the input only once is doable, but very ugly. Furthermore, it'd be nice
@@ -199,26 +199,43 @@ SkStream* handleType1Stream(SkStream* srcStream, size_t* headerLen,
     SkAutoDataUnref aud(data);
 
     if (parsePFB(src, srcLen, headerLen, dataLen, trailerLen)) {
-        SkMemoryStream* result =
-            new SkMemoryStream(*headerLen + *dataLen + *trailerLen);
-        memcpy((char*)result->getAtPos(), src + 6, *headerLen);
-        result->seek(*headerLen);
-        memcpy((char*)result->getAtPos(), src + 6 + *headerLen + 6, *dataLen);
-        result->seek(*headerLen + *dataLen);
-        memcpy((char*)result->getAtPos(), src + 6 + *headerLen + 6 + *dataLen,
-               *trailerLen);
-        result->rewind();
-        return result;
+        static const int kPFBSectionHeaderLength = 6;
+        const size_t length = *headerLen + *dataLen + *trailerLen;
+        SkASSERT(length > 0);
+        SkASSERT(length + (2 * kPFBSectionHeaderLength) <= srcLen);
+
+        SkAutoTMalloc<uint8_t> buffer(length);
+
+        const uint8_t* const srcHeader = src + kPFBSectionHeaderLength;
+        // There is a six-byte section header before header and data
+        // (but not trailer) that we're not going to copy.
+        const uint8_t* const srcData
+            = srcHeader + *headerLen + kPFBSectionHeaderLength;
+        const uint8_t* const srcTrailer = srcData + *headerLen;
+
+        uint8_t* const resultHeader = buffer.get();
+        uint8_t* const resultData = resultHeader + *headerLen;
+        uint8_t* const resultTrailer = resultData + *dataLen;
+
+        SkASSERT(resultTrailer + *trailerLen == resultHeader + length);
+
+        memcpy(resultHeader,  srcHeader,  *headerLen);
+        memcpy(resultData,    srcData,    *dataLen);
+        memcpy(resultTrailer, srcTrailer, *trailerLen);
+
+        return SkData::NewFromMalloc(buffer.detach(), length);
     }
 
     // A PFA has to be converted for PDF.
     size_t hexDataLen;
     if (parsePFA((const char*)src, srcLen, headerLen, &hexDataLen, dataLen,
                  trailerLen)) {
-        SkMemoryStream* result =
-            new SkMemoryStream(*headerLen + *dataLen + *trailerLen);
-        memcpy((char*)result->getAtPos(), src, *headerLen);
-        result->seek(*headerLen);
+        const size_t length = *headerLen + *dataLen + *trailerLen;
+        SkASSERT(length > 0);
+        SkAutoTMalloc<uint8_t> buffer(length);
+
+        memcpy(buffer.get(), src, *headerLen);
+        uint8_t* const resultData = &(buffer[*headerLen]);
 
         const uint8_t* hexData = src + *headerLen;
         const uint8_t* trailer = hexData + hexDataLen;
@@ -236,21 +253,19 @@ SkStream* handleType1Stream(SkStream* srcStream, size_t* headerLen,
             } else {
                 dataByte |= curNibble;
                 highNibble = true;
-                ((char *)result->getAtPos())[outputOffset++] = dataByte;
+                resultData[outputOffset++] = dataByte;
             }
         }
         if (!highNibble) {
-            ((char *)result->getAtPos())[outputOffset++] = dataByte;
+            resultData[outputOffset++] = dataByte;
         }
         SkASSERT(outputOffset == *dataLen);
-        result->seek(*headerLen + outputOffset);
 
-        memcpy((char *)result->getAtPos(), src + *headerLen + hexDataLen,
-               *trailerLen);
-        result->rewind();
-        return result;
-    }
+        uint8_t* const resultTrailer = &(buffer[*headerLen + outputOffset]);
+        memcpy(resultTrailer, src + *headerLen + hexDataLen, *trailerLen);
 
+        return SkData::NewFromMalloc(buffer.detach(), length);
+    }
     return NULL;
 }
 
@@ -556,9 +571,8 @@ static SkPDFStream* generate_tounicode_cmap(
     append_cmap_sections(glyphToUnicode, subset, &cmap, multiByteGlyphs,
                          firstGlyphID, lastGlyphID);
     append_cmap_footer(&cmap);
-    SkAutoTUnref<SkMemoryStream> cmapStream(new SkMemoryStream());
-    cmapStream->setData(cmap.copyToData())->unref();
-    return new SkPDFStream(cmapStream.get());
+    SkAutoTUnref<SkData> cmapData(cmap.copyToData());
+    return new SkPDFStream(cmapData.get());
 }
 
 #if defined (SK_SFNTLY_SUBSETTER)
@@ -574,6 +588,7 @@ static size_t get_subset_font_stream(const char* fontName,
                                      SkPDFStream** fontStream) {
     int ttcIndex;
     SkAutoTUnref<SkStream> fontData(typeface->openStream(&ttcIndex));
+    SkASSERT(fontData.get());
 
     size_t fontSize = fontData->getLength();
 
@@ -1295,7 +1310,7 @@ bool SkPDFType1Font::addFontDescriptor(int16_t defaultWidth) {
     size_t data SK_INIT_TO_AVOID_WARNING;
     size_t trailer SK_INIT_TO_AVOID_WARNING;
     SkAutoTUnref<SkStream> rawFontData(typeface()->openStream(&ttcIndex));
-    SkStream* fontData = handleType1Stream(rawFontData.get(), &header, &data,
+    SkData* fontData = handle_type1_stream(rawFontData.get(), &header, &data,
                                            &trailer);
     if (fontData == NULL) {
         return false;
diff --git a/src/pdf/SkPDFGraphicState.cpp b/src/pdf/SkPDFGraphicState.cpp
index 1b495341b9..fa3baaf95e 100644
--- a/src/pdf/SkPDFGraphicState.cpp
+++ b/src/pdf/SkPDFGraphicState.cpp
@@ -5,10 +5,10 @@
  * found in the LICENSE file.
  */
 
+#include "SkData.h"
 #include "SkPDFFormXObject.h"
 #include "SkPDFGraphicState.h"
 #include "SkPDFUtils.h"
-#include "SkStream.h"
 #include "SkTypes.h"
 
 static const char* blend_mode_from_xfermode(SkXfermode::Mode mode) {
@@ -121,8 +121,9 @@ SkPDFObject* SkPDFGraphicState::GetInvertFunction() {
         domainAndRange->appendInt(1);
 
         static const char psInvert[] = "{1 exch sub}";
-        SkAutoTUnref<SkMemoryStream> psInvertStream(
-            new SkMemoryStream(&psInvert, strlen(psInvert), true));
+        // Do not copy the trailing '\0' into the SkData.
+        SkAutoTUnref<SkData> psInvertStream(
+                SkData::NewWithCopy(psInvert, strlen(psInvert)));
 
         invertFunction = new SkPDFStream(psInvertStream.get());
         invertFunction->insertInt("FunctionType", 4);
diff --git a/src/pdf/SkPDFImage.cpp b/src/pdf/SkPDFImage.cpp
index 77fd84eff9..7e17f98a15 100644
--- a/src/pdf/SkPDFImage.cpp
+++ b/src/pdf/SkPDFImage.cpp
@@ -512,7 +512,7 @@ SkPDFImage::SkPDFImage(SkStream* stream,
     }
 
     if (stream != NULL) {
-        setData(stream);
+        this->setData(stream);
         fStreamValid = true;
     } else {
         fStreamValid = false;
@@ -598,13 +598,11 @@ bool SkPDFImage::populate(SkPDFCatalog* catalog) {
             SkAutoTUnref<SkData> data(fEncoder(&pixelRefOffset, subset));
             if (data.get() && data->size() < get_uncompressed_size(fBitmap,
                                                                    fSrcRect)) {
-                SkAutoTUnref<SkStream> stream(SkNEW_ARGS(SkMemoryStream,
-                                                         (data)));
-                setData(stream.get());
+                this->setData(data.get());
 
                 insertName("Filter", "DCTDecode");
                 insertInt("ColorTransform", kNoColorTransform);
-                insertInt("Length", getData()->getLength());
+                insertInt("Length", this->dataSize());
                 setState(kCompressed_State);
                 return true;
             }
@@ -613,7 +611,7 @@ bool SkPDFImage::populate(SkPDFCatalog* catalog) {
         if (!fStreamValid) {
             SkAutoTUnref<SkStream> stream(
                     extract_image_data(fBitmap, fSrcRect, fIsAlpha, NULL));
-            setData(stream);
+            this->setData(stream);
             fStreamValid = true;
         }
         return INHERITED::populate(catalog);
diff --git a/src/pdf/SkPDFPage.cpp b/src/pdf/SkPDFPage.cpp
index 8961d2f3d3..cfb6790876 100644
--- a/src/pdf/SkPDFPage.cpp
+++ b/src/pdf/SkPDFPage.cpp
@@ -7,11 +7,11 @@
  */
 
 
+#include "SkData.h"
 #include "SkPDFCatalog.h"
 #include "SkPDFDevice.h"
 #include "SkPDFPage.h"
 #include "SkPDFResourceDict.h"
-#include "SkStream.h"
 
 SkPDFPage::SkPDFPage(SkPDFDevice* content)
     : SkPDFDict("Page"),
@@ -36,7 +36,7 @@ void SkPDFPage::finalizePage(SkPDFCatalog* catalog, bool firstPage,
             }
         }
 
-        SkAutoTUnref<SkStream> content(fDevice->content());
+        SkAutoTUnref<SkData> content(fDevice->copyContentToData());
         fContentStream.reset(new SkPDFStream(content.get()));
         insert("Contents", new SkPDFObjRef(fContentStream.get()))->unref();
     }
diff --git a/src/pdf/SkPDFStream.cpp b/src/pdf/SkPDFStream.cpp
index 815ef481ee..8715d9376b 100644
--- a/src/pdf/SkPDFStream.cpp
+++ b/src/pdf/SkPDFStream.cpp
@@ -12,6 +12,7 @@
 #include "SkPDFCatalog.h"
 #include "SkPDFStream.h"
 #include "SkStream.h"
+#include "SkStreamPriv.h"
 
 static bool skip_compression(SkPDFCatalog* catalog) {
     return SkToBool(catalog->getDocumentFlags() &
@@ -19,17 +20,17 @@ static bool skip_compression(SkPDFCatalog* catalog) {
 }
 
 SkPDFStream::SkPDFStream(SkStream* stream) : fState(kUnused_State) {
-    setData(stream);
+    this->setData(stream);
 }
 
 SkPDFStream::SkPDFStream(SkData* data) : fState(kUnused_State) {
-    setData(data);
+    this->setData(data);
 }
 
 SkPDFStream::SkPDFStream(const SkPDFStream& pdfStream)
         : SkPDFDict(),
           fState(kUnused_State) {
-    setData(pdfStream.fData.get());
+    this->setData(pdfStream.fData.get());
     bool removeLength = true;
     // Don't uncompress an already compressed stream, but we could.
     if (pdfStream.fState == kCompressed_State) {
@@ -49,14 +50,16 @@ void SkPDFStream::emitObject(SkWStream* stream, SkPDFCatalog* catalog,
     if (indirect) {
         return emitIndirectObject(stream, catalog);
     }
+    SkAutoMutexAcquire lock(fMutex);  // multiple threads could be calling emit
     if (!this->populate(catalog)) {
         return fSubstitute->emitObject(stream, catalog, indirect);
     }
 
     this->INHERITED::emitObject(stream, catalog, false);
     stream->writeText(" stream\n");
-    stream->writeStream(fData.get(), fData->getLength());
-    fData->rewind();
+    if (fData.get()) {
+        stream->write(fData->data(), fData->size());
+    }
     stream->writeText("\nendstream");
 }
 
@@ -64,30 +67,34 @@ size_t SkPDFStream::getOutputSize(SkPDFCatalog* catalog, bool indirect) {
     if (indirect) {
         return getIndirectOutputSize(catalog);
     }
+    SkAutoMutexAcquire lock(fMutex);  // multiple threads could be calling emit
     if (!this->populate(catalog)) {
         return fSubstitute->getOutputSize(catalog, indirect);
     }
 
     return this->INHERITED::getOutputSize(catalog, false) +
-        strlen(" stream\n\nendstream") + fData->getLength();
+        strlen(" stream\n\nendstream") + this->dataSize();
 }
 
 SkPDFStream::SkPDFStream() : fState(kUnused_State) {}
 
 void SkPDFStream::setData(SkData* data) {
-    SkMemoryStream* stream = new SkMemoryStream;
-    stream->setData(data);
-    fData.reset(stream);  // Transfer ownership.
+    fData.reset(SkSafeRef(data));
 }
 
 void SkPDFStream::setData(SkStream* stream) {
     // Code assumes that the stream starts at the beginning and is rewindable.
     if (stream) {
         SkASSERT(stream->getPosition() == 0);
-        SkASSERT(stream->rewind());
+        fData.reset(SkCopyStreamToData(stream));
+        SkAssertResult(stream->rewind());
+    } else {
+        fData.reset(NULL);
     }
-    fData.reset(stream);
-    SkSafeRef(stream);
+}
+
+size_t SkPDFStream::dataSize() const {
+    return fData.get() ? fData->size() : 0;
 }
 
 bool SkPDFStream::populate(SkPDFCatalog* catalog) {
@@ -96,17 +103,15 @@ bool SkPDFStream::populate(SkPDFCatalog* catalog) {
             SkDynamicMemoryWStream compressedData;
 
             SkAssertResult(SkFlate::Deflate(fData.get(), &compressedData));
-            if (compressedData.getOffset() < fData->getLength()) {
-                SkMemoryStream* stream = new SkMemoryStream;
-                stream->setData(compressedData.copyToData())->unref();
-                fData.reset(stream);  // Transfer ownership.
+            if (compressedData.getOffset() < this->dataSize()) {
+                fData.reset(compressedData.copyToData());
                 insertName("Filter", "FlateDecode");
             }
             fState = kCompressed_State;
         } else {
             fState = kNoCompression_State;
         }
-        insertInt("Length", fData->getLength());
+        insertInt("Length", this->dataSize());
     } else if (fState == kNoCompression_State && !skip_compression(catalog) &&
                SkFlate::HaveFlate()) {
         if (!fSubstitute.get()) {
diff --git a/src/pdf/SkPDFStream.h b/src/pdf/SkPDFStream.h
index 636ea984e3..3a84068e3d 100644
--- a/src/pdf/SkPDFStream.h
+++ b/src/pdf/SkPDFStream.h
@@ -38,7 +38,8 @@ public:
 
     virtual ~SkPDFStream();
 
-    // The SkPDFObject interface.
+    // The SkPDFObject interface.  These two methods use a mutex to
+    // allow multiple threads to call at the same time.
     virtual void emitObject(SkWStream* stream, SkPDFCatalog* catalog,
                             bool indirect);
     virtual size_t getOutputSize(SkPDFCatalog* catalog, bool indirect);
@@ -69,22 +70,22 @@ protected:
         fSubstitute.reset(stream);
     }
 
-    SkPDFStream* getSubstitute() {
+    SkPDFStream* getSubstitute() const {
         return fSubstitute.get();
     }
 
     void setData(SkData* data);
     void setData(SkStream* stream);
 
-    SkStream* getData() {
-        return fData.get();
-    }
+    size_t dataSize() const;
+
+    SkData* getData() const { return fData.get(); }
 
     void setState(State state) {
         fState = state;
     }
 
-    State getState() {
+    State getState() const {
         return fState;
     }
 
@@ -92,8 +93,10 @@ private:
     // Indicates what form (or if) the stream has been requested.
     State fState;
 
-    // TODO(vandebo): Use SkData (after removing deprecated constructor).
-    SkAutoTUnref<SkStream> fData;
+    // Mutex guards fState, fData, and fSubstitute in public interface.
+    SkMutex fMutex;
+
+    SkAutoTUnref<SkData> fData;
     SkAutoTUnref<SkPDFStream> fSubstitute;
 
     typedef SkPDFDict INHERITED;
diff --git a/src/ports/SkImageDecoder_CG.cpp b/src/ports/SkImageDecoder_CG.cpp
index 8545ac8678..8bf30d2cc9 100644
--- a/src/ports/SkImageDecoder_CG.cpp
+++ b/src/ports/SkImageDecoder_CG.cpp
@@ -11,7 +11,7 @@
 #include "SkImageEncoder.h"
 #include "SkMovie.h"
 #include "SkStream.h"
-#include "SkStreamHelpers.h"
+#include "SkStreamPriv.h"
 #include "SkTemplates.h"
 #include "SkUnPreMultiply.h"
 
@@ -32,7 +32,7 @@ static void malloc_release_proc(void* info, const void* data, size_t size) {
 static CGDataProviderRef SkStreamToDataProvider(SkStream* stream) {
     // TODO: use callbacks, so we don't have to load all the data into RAM
     SkAutoMalloc storage;
-    const size_t len = CopyStreamToStorage(&storage, stream);
+    const size_t len = SkCopyStreamToStorage(&storage, stream);
     void* data = storage.detach();
 
     return CGDataProviderCreateWithData(data, data, len, malloc_release_proc);