From e17fb67b08077a0d2af273a3b8f49445bb7b6fb2 Mon Sep 17 00:00:00 2001 From: David Rose Date: Wed, 12 Nov 2008 18:30:59 +0000 Subject: [PATCH] streamWrapper --- dtool/src/prc/Sources.pp | 10 +- dtool/src/prc/prc_composite2.cxx | 1 + dtool/src/prc/streamWrapper.I | 208 +++++++++++++++++++++ dtool/src/prc/streamWrapper.cxx | 238 ++++++++++++++++++++++++ dtool/src/prc/streamWrapper.h | 122 ++++++++++++ panda/src/express/config_express.N | 5 + panda/src/express/config_express.cxx | 1 + panda/src/express/multifile.I | 2 +- panda/src/express/multifile.cxx | 113 +++++++---- panda/src/express/multifile.h | 7 +- panda/src/express/patchfile.cxx | 33 ++-- panda/src/express/patchfile.h | 5 +- panda/src/express/subStream.I | 4 +- panda/src/express/subStream.h | 5 +- panda/src/express/subStreamBuf.cxx | 18 +- panda/src/express/subStreamBuf.h | 7 +- panda/src/express/virtualFileSystem.cxx | 7 +- 17 files changed, 705 insertions(+), 81 deletions(-) create mode 100644 dtool/src/prc/streamWrapper.I create mode 100644 dtool/src/prc/streamWrapper.cxx create mode 100644 dtool/src/prc/streamWrapper.h diff --git a/dtool/src/prc/Sources.pp b/dtool/src/prc/Sources.pp index bb03164559..cac6f51db1 100644 --- a/dtool/src/prc/Sources.pp +++ b/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 diff --git a/dtool/src/prc/prc_composite2.cxx b/dtool/src/prc/prc_composite2.cxx index d41ef9fb4b..be7e26b368 100644 --- a/dtool/src/prc/prc_composite2.cxx +++ b/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" diff --git a/dtool/src/prc/streamWrapper.I b/dtool/src/prc/streamWrapper.I new file mode 100644 index 0000000000..d1e4e5a8fc --- /dev/null +++ b/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; +} diff --git a/dtool/src/prc/streamWrapper.cxx b/dtool/src/prc/streamWrapper.cxx new file mode 100644 index 0000000000..bc5ca10f7f --- /dev/null +++ b/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 + } +} diff --git a/dtool/src/prc/streamWrapper.h b/dtool/src/prc/streamWrapper.h new file mode 100644 index 0000000000..2e603162e7 --- /dev/null +++ b/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 diff --git a/panda/src/express/config_express.N b/panda/src/express/config_express.N index 507fce0ad0..9b381356eb 100644 --- a/panda/src/express/config_express.N +++ b/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 forcetype ConstPointerToArray renametype PointerToArray PTAUchar diff --git a/panda/src/express/config_express.cxx b/panda/src/express/config_express.cxx index bc9cabf057..dcddcc9aec 100644 --- a/panda/src/express/config_express.cxx +++ b/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, ""); diff --git a/panda/src/express/multifile.I b/panda/src/express/multifile.I index a3797862bd..fb0d1cd504 100644 --- a/panda/src/express/multifile.I +++ b/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); } //////////////////////////////////////////////////////////////////// diff --git a/panda/src/express/multifile.cxx b/panda/src/express/multifile.cxx index 74ffa1e5d1..6b2f9b50e4 100644 --- a/panda/src/express/multifile.cxx +++ b/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 ©) { +Multifile(const Multifile ©) : + _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 &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) { diff --git a/panda/src/express/multifile.h b/panda/src/express/multifile.h index c4b5c6d17d..d28cc084a4 100644 --- a/panda/src/express/multifile.h +++ b/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; diff --git a/panda/src/express/patchfile.cxx b/panda/src/express/patchfile.cxx index 4b15441e75..353f19840e 100644 --- a/panda/src/express/patchfile.cxx +++ b/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; diff --git a/panda/src/express/patchfile.h b/panda/src/express/patchfile.h index e4e767b818..083bacbe4d 100644 --- a/panda/src/express/patchfile.h +++ b/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 @@ -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; diff --git a/panda/src/express/subStream.I b/panda/src/express/subStream.I index bf4285a41d..702e9a6211 100644 --- a/panda/src/express/subStream.I +++ b/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; diff --git a/panda/src/express/subStream.h b/panda/src/express/subStream.h index dc1df32415..ee7c314569 100644 --- a/panda/src/express/subStream.h +++ b/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: diff --git a/panda/src/express/subStreamBuf.cxx b/panda/src/express/subStreamBuf.cxx index fb1a5c44bc..6a1268555b 100644 --- a/panda/src/express/subStreamBuf.cxx +++ b/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. diff --git a/panda/src/express/subStreamBuf.h b/panda/src/express/subStreamBuf.h index e24e35194d..05d7e38067 100644 --- a/panda/src/express/subStreamBuf.h +++ b/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 diff --git a/panda/src/express/virtualFileSystem.cxx b/panda/src/express/virtualFileSystem.cxx index 50ff3e9818..0bd313f87c 100644 --- a/panda/src/express/virtualFileSystem.cxx +++ b/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; }