streamWrapper

dtool/src/prc/Sources.pp

@@ -36,7 +36,9 @@
     notifySeverity.h \
     prcKeyRegistry.h \
     reversedNumericData.I reversedNumericData.h \
-    streamReader.I streamReader.h streamWriter.I streamWriter.h
+    streamReader.I streamReader.h \
+    streamWrapper.I streamWrapper.h \
+    streamWriter.I streamWriter.h
   
   #define INCLUDED_SOURCES \
     config_prc.cxx \
@@ -65,7 +67,7 @@
     notifySeverity.cxx \
     prcKeyRegistry.cxx \
     reversedNumericData.cxx \
-    streamReader.cxx streamWriter.cxx
+    streamReader.cxx streamWrapper.cxx streamWriter.cxx
   
   #define INSTALL_HEADERS \
     bigEndian.h \
@@ -97,7 +99,9 @@
     notifySeverity.h \
     prcKeyRegistry.I prcKeyRegistry.h \
     reversedNumericData.I reversedNumericData.h \
-    streamReader.I streamReader.h streamWriter.I streamWriter.h
+    streamReader.I streamReader.h \
+    streamWrapper.I streamWrapper.h \
+    streamWriter.I streamWriter.h
 
 
 #end lib_target

dtool/src/prc/prc_composite2.cxx

@@ -11,4 +11,5 @@
 #include "prcKeyRegistry.cxx"
 #include "reversedNumericData.cxx"
 #include "streamReader.cxx"
+#include "streamWrapper.cxx"
 #include "streamWriter.cxx"

dtool/src/prc/streamWrapper.I

@@ -0,0 +1,208 @@
+// Filename: streamWrapper.I
+// Created by:  drose (11Nov08)
+//
+////////////////////////////////////////////////////////////////////
+//
+// PANDA 3D SOFTWARE
+// Copyright (c) Carnegie Mellon University.  All rights reserved.
+//
+// All use of this software is subject to the terms of the revised BSD
+// license.  You should have received a copy of this license along
+// with this source code in a file named "LICENSE."
+//
+////////////////////////////////////////////////////////////////////
+
+
+////////////////////////////////////////////////////////////////////
+//     Function: StreamWrapperBase::Constructor
+//       Access: Protected
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE StreamWrapperBase::
+StreamWrapperBase() {
+#ifdef SIMPLE_THREADS
+  _lock_flag = false;
+#endif
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: StreamWrapperBase::acquire
+//       Access: Published
+//  Description: Acquires the internal lock.  
+//
+//               User code should call this to take temporary
+//               possession of the stream and perform direct I/O
+//               operations on it, for instance to make several
+//               sequential atomic reads.  You may not call any of the
+//               StreamWrapper methods while the lock is held, other
+//               than release().
+//
+//               Use with extreme caution!  This is a very low-level,
+//               non-recursive lock.  You must call acquire() only
+//               once, and you must later call release() exactly once.
+//               Failing to do so may result in a hard deadlock with
+//               no available debugging features.
+////////////////////////////////////////////////////////////////////
+INLINE void StreamWrapperBase::
+acquire() {
+  _lock.acquire();
+#ifdef SIMPLE_THREADS
+  while (_lock_flag) {
+    thread_yield();
+  }
+  _lock_flag = true;
+#endif
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: StreamWrapperBase::release
+//       Access: Published
+//  Description: Releases the internal lock.  Must be called exactly
+//               once following a call to acquire().  See the cautions
+//               with acquire().
+////////////////////////////////////////////////////////////////////
+INLINE void StreamWrapperBase::
+release() {
+#ifdef SIMPLE_THREADS
+  assert(_lock_flag);
+  _lock_flag = false;
+#endif
+  _lock.release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::Constructor
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE IStreamWrapper::
+IStreamWrapper(istream *stream, bool owns_pointer) :
+  _istream(stream),
+  _owns_pointer(owns_pointer)
+{
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::Constructor
+//       Access: Published
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE IStreamWrapper::
+IStreamWrapper(istream &stream) :
+  _istream(&stream),
+  _owns_pointer(false)
+{
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::get_istream
+//       Access: Published
+//  Description: Returns the istream this object is wrapping.
+////////////////////////////////////////////////////////////////////
+INLINE istream *IStreamWrapper::
+get_istream() const {
+  return _istream;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::get
+//       Access: Public
+//  Description: Atomically reads a single character from the stream.
+////////////////////////////////////////////////////////////////////
+INLINE int IStreamWrapper::
+get() {
+  int result;
+  acquire();
+  result = _istream->get();
+  release();
+  return result;
+}
+
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::Constructor
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE OStreamWrapper::
+OStreamWrapper(ostream *stream, bool owns_pointer) :
+  _ostream(stream),
+  _owns_pointer(owns_pointer)
+{
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::Constructor
+//       Access: Published
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE OStreamWrapper::
+OStreamWrapper(ostream &stream) :
+  _ostream(&stream),
+  _owns_pointer(false)
+{
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::get_ostream
+//       Access: Published
+//  Description: Returns the ostream this object is wrapping.
+////////////////////////////////////////////////////////////////////
+INLINE ostream *OStreamWrapper::
+get_ostream() const {
+  return _ostream;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::put
+//       Access: Public
+//  Description: Atomically writes a single character to the stream.
+//               Returns true on success, false on failure.
+////////////////////////////////////////////////////////////////////
+INLINE bool OStreamWrapper::
+put(char c) {
+  bool success;
+  acquire();
+  _ostream->put(c);
+  success = !_ostream->bad();
+  release();
+  return success;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: StreamWrapper::Constructor
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE StreamWrapper::
+StreamWrapper(iostream *stream, bool owns_pointer) :
+  IStreamWrapper(stream, false),
+  OStreamWrapper(stream, false),
+  _iostream(stream),
+  _owns_pointer(owns_pointer)
+{
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: StreamWrapper::Constructor
+//       Access: Published
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE StreamWrapper::
+StreamWrapper(iostream &stream) :
+  IStreamWrapper(&stream, false),
+  OStreamWrapper(&stream, false),
+  _iostream(&stream),
+  _owns_pointer(false)
+{
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: StreamWrapper::get_iostream
+//       Access: Published
+//  Description: Returns the iostream this object is wrapping.
+////////////////////////////////////////////////////////////////////
+INLINE iostream *StreamWrapper::
+get_iostream() const {
+  return _iostream;
+}

dtool/src/prc/streamWrapper.cxx

@@ -0,0 +1,238 @@
+// Filename: streamWrapper.cxx
+// Created by:  drose (11Nov08)
+//
+////////////////////////////////////////////////////////////////////
+//
+// PANDA 3D SOFTWARE
+// Copyright (c) Carnegie Mellon University.  All rights reserved.
+//
+// All use of this software is subject to the terms of the revised BSD
+// license.  You should have received a copy of this license along
+// with this source code in a file named "LICENSE."
+//
+////////////////////////////////////////////////////////////////////
+
+#include "streamWrapper.h"
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::Destructor
+//       Access: Published
+//  Description: 
+////////////////////////////////////////////////////////////////////
+IStreamWrapper::
+~IStreamWrapper() {
+  if (_owns_pointer) {
+    // For some reason--compiler bug in gcc 3.2?--explicitly deleting
+    // the stream pointer does not call the appropriate global delete
+    // function; instead apparently calling the system delete
+    // function.  So we call the delete function by hand instead.
+#if !defined(WIN32_VC) && !defined(USE_MEMORY_NOWRAPPERS) && defined(REDEFINE_GLOBAL_OPERATOR_NEW)
+    _istream->~istream();
+    (*global_operator_delete)(_istream);
+#else
+    delete _istream;
+#endif
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::read
+//       Access: Public
+//  Description: Atomically reads a number of bytes from the stream,
+//               without error detection.  If fewer bytes than
+//               requested are read, quietly fills the remaining bytes
+//               with 0.
+////////////////////////////////////////////////////////////////////
+void IStreamWrapper::
+read(char *buffer, streamsize num_bytes) {
+  acquire();
+  _istream->clear();
+  _istream->read(buffer, num_bytes);
+  streamsize read_bytes = _istream->gcount();
+  while (read_bytes < num_bytes) {
+    // Fewer bytes than expected were read.  Maybe more will be
+    // coming later.
+    release();
+    thread_yield();
+    acquire();
+
+    _istream->read(buffer + read_bytes, num_bytes - read_bytes);
+    streamsize this_read_bytes = _istream->gcount();
+    assert(this_read_bytes <= num_bytes - read_bytes);
+    read_bytes += this_read_bytes;
+
+    if (this_read_bytes == 0) {
+      // No, don't expect any more.
+      memset(buffer + read_bytes, 0, num_bytes - read_bytes);
+      break;
+    }
+  }
+  assert(read_bytes <= num_bytes);
+  release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::read
+//       Access: Public
+//  Description: Atomically reads a number of bytes from the stream.
+//               Returns the number of bytes actually read.
+////////////////////////////////////////////////////////////////////
+void IStreamWrapper::
+read(char *buffer, streamsize num_bytes, streamsize &read_bytes) {
+  acquire();
+  _istream->clear();
+  _istream->read(buffer, num_bytes);
+  read_bytes = _istream->gcount();
+  assert(read_bytes <= num_bytes);
+  release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::read
+//       Access: Public
+//  Description: Atomically reads a number of bytes from the stream.
+//               Returns the number of bytes actually read, and
+//               whether an eof condition was detected by the
+//               operation.
+////////////////////////////////////////////////////////////////////
+void IStreamWrapper::
+read(char *buffer, streamsize num_bytes, streamsize &read_bytes, bool &eof) {
+  acquire();
+  _istream->clear();
+  _istream->read(buffer, num_bytes);
+  read_bytes = _istream->gcount();
+  assert(read_bytes <= num_bytes);
+  eof = _istream->eof() || _istream->fail();
+  release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::seek_read
+//       Access: Public
+//  Description: Atomically seeks to a particular offset from the
+//               beginning of the file, and reads a number of bytes
+//               from the stream.  Returns the number of bytes
+//               actually read, and whether an eof condition was
+//               detected by the operation.
+////////////////////////////////////////////////////////////////////
+void IStreamWrapper::
+seek_read(streamsize pos, char *buffer, streamsize num_bytes, 
+          streamsize &read_bytes, bool &eof) {
+  acquire();
+  _istream->clear();
+  _istream->seekg(pos);
+  _istream->read(buffer, num_bytes);
+  read_bytes = _istream->gcount();
+  assert(read_bytes <= num_bytes);
+  eof = _istream->eof() || _istream->fail();
+  release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: IStreamWrapper::seek_gpos_eof
+//       Access: Public
+//  Description: Atomically seeks to EOF and returns the gpos there;
+//               that is, returns the file size.  Note that the EOF
+//               might have been moved in another thread by the time
+//               this method returns.
+////////////////////////////////////////////////////////////////////
+streamsize IStreamWrapper::
+seek_gpos_eof() {
+  streamsize pos;
+  acquire();
+  _istream->seekg(0, ios::end);
+  pos = _istream->tellg();
+  release();
+
+  return pos;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::Destructor
+//       Access: Published
+//  Description: 
+////////////////////////////////////////////////////////////////////
+OStreamWrapper::
+~OStreamWrapper() {
+  if (_owns_pointer) {
+    // For some reason--compiler bug in gcc 3.2?--explicitly deleting
+    // the stream pointer does not call the appropriate global delete
+    // function; instead apparently calling the system delete
+    // function.  So we call the delete function by hand instead.
+#if !defined(WIN32_VC) && !defined(USE_MEMORY_NOWRAPPERS) && defined(REDEFINE_GLOBAL_OPERATOR_NEW)
+    _ostream->~ostream();
+    (*global_operator_delete)(_ostream);
+#else
+    delete _ostream;
+#endif
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::write
+//       Access: Public
+//  Description: Atomically writes a number of bytes to the stream,
+//               without error detection.
+////////////////////////////////////////////////////////////////////
+void OStreamWrapper::
+write(const char *buffer, streamsize num_bytes) {
+  acquire();
+  _ostream->write(buffer, num_bytes);
+  release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::read
+//       Access: Public
+//  Description: Atomically writes a number of bytes to the stream.
+//               Returns whether a failure condition was detected by
+//               the operation.
+////////////////////////////////////////////////////////////////////
+void OStreamWrapper::
+write(const char *buffer, streamsize num_bytes, bool &fail) {
+  acquire();
+  _ostream->clear();
+  _ostream->write(buffer, num_bytes);
+  fail = _ostream->fail();
+  release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: OStreamWrapper::seek_write
+//       Access: Public
+//  Description: Atomically seeks to a particular offset from the
+//               beginning of the file, and writes a number of bytes
+//               to the stream.  Returns whether a failure condition
+//               was detected by the operation.
+////////////////////////////////////////////////////////////////////
+void OStreamWrapper::
+seek_write(streamsize pos, const char *buffer, streamsize num_bytes, 
+           bool &fail) {
+  acquire();
+  _ostream->clear();
+  _ostream->seekp(pos);
+  _ostream->write(buffer, num_bytes);
+  fail = _ostream->fail();
+  release();
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: StreamWrapper::Destructor
+//       Access: Published
+//  Description: 
+////////////////////////////////////////////////////////////////////
+StreamWrapper::
+~StreamWrapper() {
+  if (_owns_pointer) {
+    // For some reason--compiler bug in gcc 3.2?--explicitly deleting
+    // the stream pointer does not call the appropriate global delete
+    // function; instead apparently calling the system delete
+    // function.  So we call the delete function by hand instead.
+#if !defined(WIN32_VC) && !defined(USE_MEMORY_NOWRAPPERS) && defined(REDEFINE_GLOBAL_OPERATOR_NEW)
+    _iostream->~iostream();
+    (*global_operator_delete)(_iostream);
+#else
+    delete _iostream;
+#endif
+  }
+}

dtool/src/prc/streamWrapper.h

@@ -0,0 +1,122 @@
+// Filename: streamWrapper.h
+// Created by:  drose (11Nov08)
+//
+////////////////////////////////////////////////////////////////////
+//
+// PANDA 3D SOFTWARE
+// Copyright (c) Carnegie Mellon University.  All rights reserved.
+//
+// All use of this software is subject to the terms of the revised BSD
+// license.  You should have received a copy of this license along
+// with this source code in a file named "LICENSE."
+//
+////////////////////////////////////////////////////////////////////
+
+#ifndef STREAMWRAPPER_H
+#define STREAMWRAPPER_H
+
+#include "dtoolbase.h"
+#include "mutexImpl.h"
+
+////////////////////////////////////////////////////////////////////
+//       Class : StreamWrapperBase
+// Description : The base class for both IStreamWrapper and
+//               OStreamWrapper, this provides the common locking
+//               interface.
+////////////////////////////////////////////////////////////////////
+class EXPCL_DTOOLCONFIG StreamWrapperBase {
+protected:
+  INLINE StreamWrapperBase();
+
+PUBLISHED:
+  INLINE void acquire();
+  INLINE void release();
+
+private:
+  MutexImpl _lock;
+#ifdef SIMPLE_THREADS
+  // In the SIMPLE_THREADS case, we need to use a bool flag, because
+  // MutexImpl defines to nothing in this case--but we still need to
+  // achieve a form of locking, since I/O operations can cause the
+  // thread to swap without warning.
+  bool _lock_flag;
+#endif
+};
+
+////////////////////////////////////////////////////////////////////
+//       Class : IStreamWrapper
+// Description : This class provides a locking wrapper around an
+//               arbitrary istream pointer.  A thread may use this
+//               class to perform an atomic seek/read/gcount
+//               operation.
+////////////////////////////////////////////////////////////////////
+class EXPCL_DTOOLCONFIG IStreamWrapper : virtual public StreamWrapperBase {
+public:
+  INLINE IStreamWrapper(istream *stream, bool owns_pointer);
+PUBLISHED:
+  INLINE IStreamWrapper(istream &stream);
+  ~IStreamWrapper();
+
+  INLINE istream *get_istream() const;
+
+public:
+  void read(char *buffer, streamsize num_bytes);
+  void read(char *buffer, streamsize num_bytes, streamsize &read_bytes);
+  void read(char *buffer, streamsize num_bytes, streamsize &read_bytes, bool &eof);
+  void seek_read(streamsize pos, char *buffer, streamsize num_bytes, streamsize &read_bytes, bool &eof);
+  INLINE int get();
+  streamsize seek_gpos_eof();
+
+private:
+  istream *_istream;
+  bool _owns_pointer;
+};
+
+////////////////////////////////////////////////////////////////////
+//       Class : OStreamWrapper
+// Description : This class provides a locking wrapper around an
+//               arbitrary ostream pointer.  A thread may use this
+//               class to perform an atomic seek/write operation.
+////////////////////////////////////////////////////////////////////
+class EXPCL_DTOOLCONFIG OStreamWrapper : virtual public StreamWrapperBase {
+public:
+  INLINE OStreamWrapper(ostream *stream, bool owns_pointer);
+PUBLISHED:
+  INLINE OStreamWrapper(ostream &stream);
+  ~OStreamWrapper();
+
+  INLINE ostream *get_ostream() const;
+
+public:
+  void write(const char *buffer, streamsize num_bytes);
+  void write(const char *buffer, streamsize num_bytes, bool &fail);
+  void seek_write(streamsize pos, const char *buffer, streamsize num_bytes, bool &fail);
+  INLINE bool put(char c);
+
+private:
+  ostream *_ostream;
+  bool _owns_pointer;
+};
+
+////////////////////////////////////////////////////////////////////
+//       Class : StreamWrapper
+// Description : This class provides a locking wrapper around a
+//               combination ostream/istream pointer.
+////////////////////////////////////////////////////////////////////
+class EXPCL_DTOOLCONFIG StreamWrapper : public IStreamWrapper, public OStreamWrapper {
+public:
+  INLINE StreamWrapper(iostream *stream, bool owns_pointer);
+PUBLISHED:
+  INLINE StreamWrapper(iostream &stream);
+  ~StreamWrapper();
+
+  INLINE iostream *get_iostream() const;
+
+private:
+  iostream *_iostream;
+  bool _owns_pointer;
+};
+
+#include "streamWrapper.I"
+
+#endif

panda/src/express/config_express.N

@@ -43,6 +43,11 @@ forcetype istream
 forcetype ostream
 forcetype iostream
 
+forcetype StreamWrapperBase
+forcetype IStreamWrapper
+forcetype OStreamWrapper
+forcetype StreamWrapper
+
 forcetype PointerToArray<unsigned char>
 forcetype ConstPointerToArray<unsigned char>
 renametype PointerToArray<unsigned char> PTAUchar

panda/src/express/config_express.cxx

@@ -30,6 +30,7 @@
 #include "namable.h"
 #include "export_dtool.h"
 #include "dconfig.h"
+#include "streamWrapper.h"
 
 ConfigureDef(config_express);
 NotifyCategoryDef(express, "");

panda/src/express/multifile.I

@@ -46,7 +46,7 @@ set_multifile_name(const Filename &multifile_name) {
 ////////////////////////////////////////////////////////////////////
 INLINE bool Multifile::
 is_read_valid() const {
-  return (_read != (istream *)NULL && !_read->fail());
+  return (_read != (IStreamWrapper *)NULL);
 }
 
 ////////////////////////////////////////////////////////////////////

panda/src/express/multifile.cxx

@@ -99,8 +99,11 @@ const size_t Multifile::_encrypt_header_size = 6;
 //  Description:
 ////////////////////////////////////////////////////////////////////
 Multifile::
-Multifile() {
-  _read = (istream *)NULL;
+Multifile() :
+  _read_filew(_read_file),
+  _read_write_filew(_read_write_file)
+{
+  _read = (IStreamWrapper *)NULL;
   _write = (ostream *)NULL;
   _owns_stream = false;
   _next_index = 0;
@@ -132,7 +135,10 @@ Multifile::
 //  Description: Don't try to copy Multifiles.
 ////////////////////////////////////////////////////////////////////
 Multifile::
-Multifile(const Multifile &copy) {
+Multifile(const Multifile &copy) :
+  _read_filew(_read_file),
+  _read_write_filew(_read_write_file)
+{
   nassertv(false);
 }
 
@@ -168,7 +174,7 @@ open_read(const Filename &multifile_name) {
   }
   _timestamp = fname.get_timestamp();
   _timestamp_dirty = true;
-  _read = &_read_file;
+  _read = &_read_filew;
   _multifile_name = multifile_name;
   return read_index();
 }
@@ -186,13 +192,12 @@ open_read(const Filename &multifile_name) {
 //               function returns false.
 ////////////////////////////////////////////////////////////////////
 bool Multifile::
-open_read(istream *multifile_stream, bool owns_pointer) {
+open_read(IStreamWrapper *multifile_stream, bool owns_pointer) {
   close();
   _timestamp = time(NULL);
   _timestamp_dirty = true;
   _read = multifile_stream;
   _owns_stream = owns_pointer;
-  _read->seekg(0, ios::beg);
   return read_index();
 }
 
@@ -275,7 +280,7 @@ open_read_write(const Filename &multifile_name) {
     _timestamp = time(NULL);
   }
   _timestamp_dirty = true;
-  _read = &_read_write_file;
+  _read = &_read_write_filew;
   _write = &_read_write_file;
   _multifile_name = multifile_name;
 
@@ -304,21 +309,25 @@ open_read_write(iostream *multifile_stream, bool owns_pointer) {
   close();
   _timestamp = time(NULL);
   _timestamp_dirty = true;
-  _read = multifile_stream;
+
+  // We don't support locking when opening a file in read-write mode,
+  // because we don't bother with locking on write.  But we need to
+  // have an IStreamWrapper to assign to the _read member, so we
+  // create one on-the-fly here.
+  _read = new StreamWrapper(multifile_stream, owns_pointer);
   _write = multifile_stream;
-  _owns_stream = owns_pointer;
+  _owns_stream = true;  // Because we own the StreamWrapper, above.
   _write->seekp(0, ios::beg);
 
   // Check whether the read stream is empty.
-  _read->seekg(0, ios::end);
-  if (_read->tellg() == (streampos)0) {
+  multifile_stream->seekg(0, ios::end);
+  if (multifile_stream->tellg() == (streampos)0) {
     // The read stream is empty, which is always valid.
     return true;
   }
 
   // The read stream is not empty, so we'd better have a valid
   // Multifile.
-  _read->seekg(0, ios::beg);
   return read_index();
 }
 
@@ -340,14 +349,16 @@ close() {
   }
 
   if (_owns_stream) {
-    if (_read != (istream *)NULL) {
-      VirtualFileSystem::close_read_file(_read);
+    // We prefer to delete the IStreamWrapper over the ostream, if
+    // possible.
+    if (_read != (IStreamWrapper *)NULL) {
+      delete _read;
     } else if (_write != (ostream *)NULL) {
       delete _write;
     }
   }
 
-  _read = (istream *)NULL;
+  _read = (IStreamWrapper *)NULL;
   _write = (ostream *)NULL;
   _owns_stream = false;
   _next_index = 0;
@@ -630,8 +641,10 @@ flush() {
     // All right, now write out each subfile's data.
     for (pi = _new_subfiles.begin(); pi != _new_subfiles.end(); ++pi) {
       Subfile *subfile = (*pi);
-      _next_index = subfile->write_data(*_write, _read, _next_index,
-                                        this);
+      _read->acquire();
+      _next_index = subfile->write_data(*_write, _read->get_istream(),
+                                        _next_index, this);
+      _read->release();
       nassertr(_next_index == _write->tellp(), false);
       _next_index = pad_to_streampos(_next_index);
       if (subfile->is_data_invalid()) {
@@ -1167,10 +1180,15 @@ extract_subfile_to(int index, ostream &out) {
     return false;
   }
 
-  int byte = in->get();
-  while (!in->fail() && !in->eof()) {
-    out.put(byte);
-    byte = in->get();
+  static const size_t buffer_size = 1024;
+  char buffer[buffer_size];
+
+  in->read(buffer, buffer_size);
+  size_t count = in->gcount();
+  while (count != 0) {
+    out.write(buffer, count);
+    in->read(buffer, buffer_size);
+    count = in->gcount();
   }
 
   bool failed = (in->fail() && !in->eof());
@@ -1315,7 +1333,6 @@ read_subfile(int index, pvector<unsigned char> &result) {
   in->read(buffer, buffer_size);
   size_t count = in->gcount();
   while (count != 0) {
-    thread_consider_yield();
     result.insert(result.end(), buffer, buffer + count);
     in->read(buffer, buffer_size);
     count = in->gcount();
@@ -1460,32 +1477,43 @@ clear_subfiles() {
 ////////////////////////////////////////////////////////////////////
 bool Multifile::
 read_index() {
-  nassertr(_read != (istream *)NULL, false);
+  nassertr(_read != (IStreamWrapper *)NULL, false);
+
+  // We acquire the IStreamWrapper lock for the duration of this
+  // method.
+  _read->acquire();
+  istream *read = _read->get_istream();
 
   char this_header[_header_size];
-  _read->read(this_header, _header_size);
-  if (_read->fail() || _read->gcount() != (unsigned)_header_size) {
+  read->seekg(0);
+  read->read(this_header, _header_size);
+  if (read->fail() || read->gcount() != (unsigned)_header_size) {
     express_cat.info()
       << "Unable to read Multifile header " << _multifile_name << ".\n";
+    _read->release();
     close();
     return false;
   }
   if (memcmp(this_header, _header, _header_size) != 0) {
     express_cat.info()
       << _multifile_name << " is not a Multifile.\n";
+    _read->release();
+    close();
     return false;
   }
 
   // Now get the version numbers out.
-  StreamReader reader(_read, false);
+  StreamReader reader(read, false);
   _file_major_ver = reader.get_int16();
   _file_minor_ver = reader.get_int16();
   _scale_factor = reader.get_uint32();
   _new_scale_factor = _scale_factor;
 
-  if (_read->eof() || _read->fail()) {
+  if (read->eof() || read->fail()) {
     express_cat.info()
       << _multifile_name << " header is truncated.\n";
+    _read->release();
+    close();
     return false;
   }
 
@@ -1496,6 +1524,8 @@ read_index() {
       << _multifile_name << " has version " << _file_major_ver << "."
       << _file_minor_ver << ", expecting version " 
       << _current_major_ver << "." << _current_minor_ver << ".\n";
+    _read->release();
+    close();
     return false;
   }
 
@@ -1513,14 +1543,14 @@ read_index() {
   }
 
   // Now read the index out.
-  _next_index = _read->tellg();
+  _next_index = read->tellg();
   _next_index = normalize_streampos(_next_index);  
-  _read->seekg(_next_index);
+  read->seekg(_next_index);
   _last_index = 0;
   streampos index_forward;
 
   Subfile *subfile = new Subfile;
-  index_forward = subfile->read_index(*_read, _next_index, this);
+  index_forward = subfile->read_index(*read, _next_index, this);
   while (index_forward != (streampos)0) {
     _last_index = _next_index;
     if (subfile->is_deleted()) {
@@ -1530,19 +1560,20 @@ read_index() {
     } else {
       _subfiles.push_back(subfile);
     }
-    if (index_forward != normalize_streampos(_read->tellg())) {
+    if (index_forward != normalize_streampos(read->tellg())) {
       // If the index entries don't follow exactly sequentially, the
       // file ought to be repacked.
       _needs_repack = true;
     }
-    _read->seekg(index_forward);
+    read->seekg(index_forward);
     _next_index = index_forward;
     subfile = new Subfile;
-    index_forward = subfile->read_index(*_read, _next_index, this);
+    index_forward = subfile->read_index(*read, _next_index, this);
   }
   if (subfile->is_index_invalid()) {
     express_cat.info()
       << "Error reading index for " << _multifile_name << ".\n";
+    _read->release();
     close();
     delete subfile;
     return false;
@@ -1568,6 +1599,7 @@ read_index() {
   }
 
   delete subfile;
+  _read->release();
   return true;
 }  
 
@@ -1842,11 +1874,16 @@ write_data(ostream &write, istream *read, streampos fpos,
     streampos write_start = fpos;
     _uncompressed_length = 0;
 
-    int byte = source->get();
-    while (!source->eof() && !source->fail()) {
-      _uncompressed_length++;
-      putter->put(byte);
-      byte = source->get();
+    static const size_t buffer_size = 1024;
+    char buffer[buffer_size];
+    
+    source->read(buffer, buffer_size);
+    size_t count = source->gcount();
+    while (count != 0) {
+      _uncompressed_length += count;
+      putter->write(buffer, count);
+      source->read(buffer, buffer_size);
+      count = source->gcount();
     }
 
     if (delete_putter) {

panda/src/express/multifile.h

@@ -18,6 +18,7 @@
 #include "pandabase.h"
 
 #include "config_express.h"
+#include "streamWrapper.h"
 #include "subStream.h"
 #include "filename.h"
 #include "ordered_vector.h"
@@ -40,7 +41,7 @@ private:
 
 PUBLISHED:
   BLOCKING bool open_read(const Filename &multifile_name);
-  BLOCKING bool open_read(istream *multifile_stream, bool owns_pointer = false);
+  BLOCKING bool open_read(IStreamWrapper *multifile_stream, bool owns_pointer = false);
   BLOCKING bool open_write(const Filename &multifile_name);
   BLOCKING bool open_write(ostream *multifile_stream, bool owns_pointer = false);
   BLOCKING bool open_read_write(const Filename &multifile_name);
@@ -164,7 +165,7 @@ private:
   PendingSubfiles _new_subfiles;
   PendingSubfiles _removed_subfiles;
 
-  istream *_read;
+  IStreamWrapper *_read;
   ostream *_write;
   bool _owns_stream;
   streampos _next_index;
@@ -181,8 +182,10 @@ private:
   string _encryption_password;
 
   pifstream _read_file;
+  IStreamWrapper _read_filew;
   pofstream _write_file;
   pfstream _read_write_file;
+  StreamWrapper _read_write_filew;
   Filename _multifile_name;
 
   int _file_major_ver;

panda/src/express/patchfile.cxx

@@ -1100,8 +1100,10 @@ compute_mf_patches(ostream &write_stream,
                    PN_uint32 offset_orig, PN_uint32 offset_new,
                    istream &stream_orig, istream &stream_new) {
   Multifile mf_orig, mf_new;
-  if (!mf_orig.open_read(&stream_orig) ||
-      !mf_new.open_read(&stream_new)) {
+  IStreamWrapper stream_origw(stream_orig);
+  IStreamWrapper stream_neww(stream_new);
+  if (!mf_orig.open_read(&stream_origw) ||
+      !mf_new.open_read(&stream_neww)) {
     express_cat.error()
       << "Input multifiles appear to be corrupt.\n";
     return false;
@@ -1116,8 +1118,8 @@ compute_mf_patches(ostream &write_stream,
   // First, compute the patch for the header / index.
 
   {
-    ISubStream index_orig(&stream_orig, 0, mf_orig.get_index_end());
-    ISubStream index_new(&stream_new, 0, mf_new.get_index_end());
+    ISubStream index_orig(&stream_origw, 0, mf_orig.get_index_end());
+    ISubStream index_new(&stream_neww, 0, mf_new.get_index_end());
     if (!do_compute_patches("", "",
                             write_stream, offset_orig, offset_new,
                             index_orig, index_new)) {
@@ -1161,8 +1163,8 @@ compute_mf_patches(ostream &write_stream,
 
       if (!patch_subfile(write_stream, offset_orig, offset_new, 
                          mf_new.get_subfile_name(ni),
-                         stream_orig, orig_start, orig_start + (streampos)orig_size,
-                         stream_new, new_start, new_start + (streampos)new_size)) {
+                         stream_origw, orig_start, orig_start + (streampos)orig_size,
+                         stream_neww, new_start, new_start + (streampos)new_size)) {
         return false;
       }
     }
@@ -1264,6 +1266,9 @@ compute_tar_patches(ostream &write_stream,
   // subfile has been removed, we simply don't add it (we'll never
   // even notice this case).
 
+  IStreamWrapper stream_origw(stream_orig);
+  IStreamWrapper stream_neww(stream_new);
+
   TarDef::const_iterator ni;
   streampos last_pos = 0;
   for (ni = tar_new.begin(); ni != tar_new.end(); ++ni) {
@@ -1295,20 +1300,20 @@ compute_tar_patches(ostream &write_stream,
       // the end of the file (possibly introduced by a tar file's
       // blocking) is the footer, which is also patched separately.
       if (!patch_subfile(write_stream, offset_orig, offset_new, "", 
-                         stream_orig, sf_orig._header_start, sf_orig._data_start,
-                         stream_new, sf_new._header_start, sf_new._data_start)) {
+                         stream_origw, sf_orig._header_start, sf_orig._data_start,
+                         stream_neww, sf_new._header_start, sf_new._data_start)) {
         return false;
       }
 
       if (!patch_subfile(write_stream, offset_orig, offset_new, sf_new._name,
-                         stream_orig, sf_orig._data_start, sf_orig._data_end,
-                         stream_new, sf_new._data_start, sf_new._data_end)) {
+                         stream_origw, sf_orig._data_start, sf_orig._data_end,
+                         stream_neww, sf_new._data_start, sf_new._data_end)) {
         return false;
       }
 
       if (!patch_subfile(write_stream, offset_orig, offset_new, "",
-                         stream_orig, sf_orig._data_end, sf_orig._end,
-                         stream_new, sf_new._data_end, sf_new._end)) {
+                         stream_origw, sf_orig._data_end, sf_orig._end,
+                         stream_neww, sf_new._data_end, sf_new._end)) {
         return false;
       }
     }
@@ -1552,8 +1557,8 @@ bool Patchfile::
 patch_subfile(ostream &write_stream, 
               PN_uint32 offset_orig, PN_uint32 offset_new,
               const Filename &filename,
-              istream &stream_orig, streampos orig_start, streampos orig_end,
-              istream &stream_new, streampos new_start, streampos new_end) {
+              IStreamWrapper &stream_orig, streampos orig_start, streampos orig_end,
+              IStreamWrapper &stream_new, streampos new_start, streampos new_end) {
   nassertr(_add_pos + _cache_add_data.size() + _cache_copy_length == offset_new + new_start, false);
 
   size_t new_size = new_end - new_start;

panda/src/express/patchfile.h

@@ -29,6 +29,7 @@
 #include "pointerTo.h"
 #include "hashVal.h" // MD5 stuff
 #include "ordered_vector.h"
+#include "streamWrapper.h"
 
 #include <algorithm>
 
@@ -138,8 +139,8 @@ private:
   bool patch_subfile(ostream &write_stream, 
                      PN_uint32 offset_orig, PN_uint32 offset_new,
                      const Filename &filename,
-                     istream &stream_orig, streampos orig_start, streampos orig_end,
-                     istream &stream_new, streampos new_start, streampos new_end);
+                     IStreamWrapper &stream_orig, streampos orig_start, streampos orig_end,
+                     IStreamWrapper &stream_new, streampos new_start, streampos new_end);
 
   static const PN_uint32 _HASH_BITS;
   static const PN_uint32 _HASHTABLESIZE;

panda/src/express/subStream.I

@@ -28,7 +28,7 @@ ISubStream() : istream(&_buf) {
 //  Description:
 ////////////////////////////////////////////////////////////////////
 INLINE ISubStream::
-ISubStream(istream *source, streampos start, streampos end) : istream(&_buf) {
+ISubStream(IStreamWrapper *source, streampos start, streampos end) : istream(&_buf) {
   open(source, start, end);
 }
 
@@ -46,7 +46,7 @@ ISubStream(istream *source, streampos start, streampos end) : istream(&_buf) {
 //               continue until the end of the source stream.
 ////////////////////////////////////////////////////////////////////
 INLINE ISubStream &ISubStream::
-open(istream *source, streampos start, streampos end) {
+open(IStreamWrapper *source, streampos start, streampos end) {
   clear((ios_iostate)0);
   _buf.open(source, start, end);
   return *this;

panda/src/express/subStream.h

@@ -17,6 +17,7 @@
 
 #include "pandabase.h"
 #include "subStreamBuf.h"
+#include "streamWrapper.h"
 
 ////////////////////////////////////////////////////////////////////
 //       Class : ISubStream
@@ -33,9 +34,9 @@
 class EXPCL_PANDAEXPRESS ISubStream : public istream {
 public:
   INLINE ISubStream();
-  INLINE ISubStream(istream *source, streampos start, streampos end);
+  INLINE ISubStream(IStreamWrapper *source, streampos start, streampos end);
 
-  INLINE ISubStream &open(istream *source, streampos start, streampos end);
+  INLINE ISubStream &open(IStreamWrapper *source, streampos start, streampos end);
   INLINE ISubStream &close();
 
 private:

panda/src/express/subStreamBuf.cxx

@@ -27,7 +27,7 @@ typedef int streamsize;
 ////////////////////////////////////////////////////////////////////
 SubStreamBuf::
 SubStreamBuf() {
-  _source = (istream *)NULL;
+  _source = (IStreamWrapper *)NULL;
 
   // _start is the streampos of the first byte of the SubStream within
   // its parent stream.
@@ -82,7 +82,7 @@ SubStreamBuf::
 //  Description:
 ////////////////////////////////////////////////////////////////////
 void SubStreamBuf::
-open(istream *source, streampos start, streampos end) {
+open(IStreamWrapper *source, streampos start, streampos end) {
   _source = source;
   _start = start;
   _end = end;
@@ -99,7 +99,7 @@ open(istream *source, streampos start, streampos end) {
 ////////////////////////////////////////////////////////////////////
 void SubStreamBuf::
 close() {
-  _source = (istream *)NULL;
+  _source = (IStreamWrapper *)NULL;
   _start = 0;
   _end = 0;
   _cur = 0;
@@ -138,10 +138,7 @@ seekoff(streamoff off, ios_seekdir dir, ios_openmode mode) {
     if (_end == (streampos)0) {
       // If the end of the file is unspecified, we have to seek to
       // find it.
-      _lock.acquire();
-      _source->seekg(off, ios::end);
-      new_pos = _source->tellg();
-      _lock.release();
+      new_pos = _source->seek_gpos_eof() + off;
 
     } else {
       new_pos = _end + off;
@@ -239,11 +236,8 @@ underflow() {
     gbump(-(int)num_bytes);
     nassertr(gptr() + num_bytes <= egptr(), EOF);
 
-    _lock.acquire();
-    _source->seekg(_cur);
-    _source->read(gptr(), num_bytes);
-    size_t read_count = _source->gcount();
-    _lock.release();
+    streamsize read_count;
+    _source->read(gptr(), num_bytes, read_count);
 
     if (read_count != num_bytes) {
       // Oops, we didn't read what we thought we would.

panda/src/express/subStreamBuf.h

@@ -16,7 +16,7 @@
 #define SUBSTREAMBUF_H
 
 #include "pandabase.h"
-#include "mutexImpl.h"
+#include "streamWrapper.h"
 
 ////////////////////////////////////////////////////////////////////
 //       Class : SubStreamBuf
@@ -27,7 +27,7 @@ public:
   SubStreamBuf();
   virtual ~SubStreamBuf();
 
-  void open(istream *source, streampos start, streampos end);
+  void open(IStreamWrapper *source, streampos start, streampos end);
   void close();
 
   virtual streampos seekoff(streamoff off, ios_seekdir dir, ios_openmode mode);
@@ -39,13 +39,12 @@ protected:
   virtual int underflow();
 
 private:
-  istream *_source;
+  IStreamWrapper *_source;
   streampos _start;
   streampos _end;
   streampos _cur;
   size_t _unused;
   char *_buffer;
-  MutexImpl _lock;
 };
 
 #endif

panda/src/express/virtualFileSystem.cxx

@@ -18,6 +18,7 @@
 #include "virtualFileMount.h"
 #include "virtualFileMountMultifile.h"
 #include "virtualFileMountSystem.h"
+#include "streamWrapper.h"
 #include "dSearchPath.h"
 #include "dcast.h"
 #include "config_express.h"
@@ -1035,7 +1036,11 @@ consider_mount_mf(const Filename &filename) {
       return false;
     }
 
-    if (!multifile->open_read(stream, true)) {
+    // Wrap a thread-safe wrapper around that stream, so multiple
+    // threads can safely read the multifile simultaneously.
+    IStreamWrapper *streamw = new IStreamWrapper(stream, true);
+
+    if (!multifile->open_read(streamw, true)) {
       // Invalid multifile.
       return false;
     }