43 template<
typename RSP>
45 const std::string &fn,
46 uint64_t relativeOffset,
52 if( me.openstage != ZipArchive::Done || !me.archive.
IsOpen() )
57 auto cditr = me.cdmap.find( fn );
58 if( cditr == me.cdmap.end() )
62 CDFH *cdfh = me.cdvec[cditr->second].get();
67 0,
"The compression algorithm is not supported!" );
76 uint64_t cdOffset = me.zip64eocd ? me.zip64eocd->cdOffset : me.eocd->cdOffset;
77 uint64_t nextRecordOffset = ( cditr->second + 1 < me.cdvec.size() ) ?
80 if( filesize == std::numeric_limits<uint32_t>::max() && cdfh->
extra )
81 filesize = cdfh->
extra->compressedSize;
84 uint64_t fileoff = nextRecordOffset - filesize - descsize;
85 uint64_t offset = fileoff + relativeOffset;
87 if( uncompressedSize == std::numeric_limits<uint32_t>::max() && cdfh->
extra )
88 uncompressedSize = cdfh->
extra->uncompressedSize;
89 uint64_t sizeTillEnd = relativeOffset > uncompressedSize ?
90 0 : uncompressedSize - relativeOffset;
91 if( size > sizeTillEnd ) size = sizeTillEnd;
96 log->
Dump(
ZipMsg,
"[%p] Reading compressed data.", &me );
98 bool empty = me.zipcache.find( fn ) == me.zipcache.end();
103 if( relativeOffset > uncompressedSize )
107 RSP *r =
new RSP( relativeOffset, 0, usrbuff );
114 uint32_t sizereq = size;
115 if( relativeOffset + size > uncompressedSize )
116 sizereq = uncompressedSize - relativeOffset;
117 cache.
QueueReq( relativeOffset, sizereq, usrbuff, usrHandler );
121 if( empty && me.buffer)
123 auto begin = me.buffer.get() + fileoff;
124 auto end = begin + filesize ;
135 uint32_t rdsize = size;
138 if( relativeOffset + size >= uncompressedSize )
143 rdsize = filesize > relativeOffset ?
144 filesize - relativeOffset :
149 if( relativeOffset + size > filesize )
150 rdsize = filesize - relativeOffset;
154 auto rdbuff = std::make_shared<ZipCache::buffer_t>( rdsize );
155 Pipeline p = XrdCl::RdWithRsp<RSP>( me.archive, offset, rdbuff->size(), rdbuff->data() ) >>
156 [relativeOffset, rdbuff, &cache, &me](
XRootDStatus &st, RSP &rsp )
159 log->
Dump(
ZipMsg,
"[%p] Read %u bytes of remote data at offset %llu.",
160 &me, rsp.GetLength(), (
unsigned long long) rsp.
GetOffset() );
161 cache.QueueRsp( st, relativeOffset, std::move( *rdbuff ) );
163 Async( std::move( p ), timeout );
170 if( me.buffer || size == 0 )
174 memcpy( usrbuff, me.buffer.get() + offset, size );
175 log->
Dump(
ZipMsg,
"[%p] Serving read from local cache.", &me );
181 RSP *rsp =
new RSP( relativeOffset, size, usrbuff );
182 ZipArchive::Schedule( usrHandler, st, rsp );
187 Pipeline p = XrdCl::RdWithRsp<RSP>( me.archive, offset, size, usrbuff ) >>
190 log->
Dump(
ZipMsg,
"[%p] Read %u bytes of remote data at "
191 "offset %llu.", &me, r.GetLength(), (
unsigned long long) r.
GetOffset() );
197 rsp =
new RSP( relativeOffset, r.GetLength(), r.GetBuffer() );
201 Async( std::move( p ), timeout );
231 XRootDStatus ZipArchive::OpenOnly(
const std::string &url,
244 archsize = info.GetSize();
245 openstage = NotParsed;
246 log->
Debug(
ZipMsg,
"[%p] Opened (only) a ZIP archive (%s).",
251 log->
Error(
ZipMsg,
"[%p] Failed to open-only a ZIP archive (%s): %s",
252 this, url.c_str(), st.
ToString().c_str() );
259 Async( std::move( open_only ), timeout );
260 return XRootDStatus();
283 if( !status.
IsOK() )
return;
285 archsize = info.GetSize();
291 log->
Dump(
ZipMsg,
"[%p] Opened a ZIP archive (file empty).",
this );
295 rdsize = ( archsize <= maxrdsz ? archsize : maxrdsz );
296 rdoff = archsize - *rdsize;
297 buffer.reset(
new char[*rdsize] );
298 rdbuff = buffer.get();
299 openstage = HaveEocdBlk;
300 log->
Dump(
ZipMsg,
"[%p] Opened a ZIP archive, reading "
301 "Central Directory at offset: %llu.",
this, (
unsigned long long) *rdoff );
308 if( !status.
IsOK() )
return;
310 const char *buff =
reinterpret_cast<char*
>( chunk.buffer );
318 const char *eocdBlock =
EOCD::Find( buff, chunk.length );
322 "End-of-central-directory signature not found." );
326 eocd.reset(
new EOCD( eocdBlock, chunk.length - uint32_t(eocdBlock - buff) ) );
327 log->
Dump(
ZipMsg,
"[%p] EOCD record parsed: %s",
this,
328 eocd->ToString().c_str() );
329 if(eocd->cdOffset > archsize || eocd->cdOffset + eocd->cdSize > archsize)
334 "End-of-central-directory signature corrupted." );
338 if( chunk.length == archsize )
342 cdoff = eocd->cdOffset;
343 orgcdsz = eocd->cdSize;
344 orgcdcnt = eocd->nbCdRec;
346 openstage = HaveCdRecords;
353 if( zip64EocdlBlock > buffer.get() )
355 uint32_t signature = to<uint32_t>( zip64EocdlBlock );
358 buff = zip64EocdlBlock;
359 openstage = HaveZip64EocdlBlk;
366 cdoff = eocd->cdOffset;
367 orgcdsz = eocd->cdSize;
368 orgcdcnt = eocd->nbCdRec;
369 rdoff = eocd->cdOffset;
370 rdsize = eocd->cdSize;
371 buffer.reset(
new char[*rdsize] );
372 rdbuff = buffer.get();
373 openstage = HaveCdRecords;
374 log->
Dump(
ZipMsg,
"[%p] Reading additional data at offset: %llu.",
375 this, (
unsigned long long) *rdoff );
379 case HaveZip64EocdlBlk:
381 std::unique_ptr<ZIP64_EOCDL> eocdl(
new ZIP64_EOCDL( buff ) );
382 log->
Dump(
ZipMsg,
"[%p] EOCDL record parsed: %s",
383 this, eocdl->ToString().c_str() );
385 if( chunk.offset > eocdl->zip64EocdOffset )
388 rdsize = archsize - eocdl->zip64EocdOffset;
389 rdoff = eocdl->zip64EocdOffset;
390 buffer.reset(
new char[*rdsize] );
391 rdbuff = buffer.get();
392 openstage = HaveZip64EocdBlk;
393 log->
Dump(
ZipMsg,
"[%p] Reading additional data at offset: %llu.",
394 this, (
unsigned long long) *rdoff );
398 buff = buffer.get() + ( eocdl->zip64EocdOffset - chunk.offset );
399 openstage = HaveZip64EocdBlk;
403 case HaveZip64EocdBlk:
405 uint32_t signature = to<uint32_t>( buff );
409 "ZIP64 End-of-central-directory signature not found." );
413 log->
Dump(
ZipMsg,
"[%p] ZIP64EOCD record parsed: %s",
414 this, zip64eocd->ToString().c_str() );
417 cdoff = zip64eocd->cdOffset;
418 orgcdsz = zip64eocd->cdSize;
419 orgcdcnt = zip64eocd->nbCdRec;
420 rdoff = zip64eocd->cdOffset;
421 rdsize = zip64eocd->cdSize;
422 buffer.reset(
new char[*rdsize] );
423 rdbuff = buffer.get();
424 openstage = HaveCdRecords;
425 log->
Dump(
ZipMsg,
"[%p] Reading additional data at offset: %llu.",
426 this, (
unsigned long long) *rdoff );
433 orgcdbuf.reserve( orgcdsz );
434 std::copy( buff, buff + orgcdsz, std::back_inserter( orgcdbuf ) );
438 std::tie( cdvec, cdmap ) =
CDFH::Parse( buff, zip64eocd->cdSize, zip64eocd->nbCdRec );
440 std::tie( cdvec, cdmap ) =
CDFH::Parse( buff, eocd->cdSize, eocd->nbCdRec );
441 log->
Dump(
ZipMsg,
"[%p] CD records parsed.",
this );
442 uint64_t sumCompSize = 0;
443 for (
auto it = cdvec.begin(); it != cdvec.end(); it++)
445 sumCompSize += (*it)->IsZIP64() ? (*it)->extra->compressedSize : (*it)->compressedSize;
446 if ((*it)->offset > archsize || (*it)->offset + (*it)->compressedSize > archsize)
449 if (sumCompSize > archsize)
455 "ZIP Central Directory corrupted." );
458 if( chunk.length != archsize ) buffer.reset();
473 log->
Debug(
ZipMsg,
"[%p] Opened a ZIP archive (%s): %s",
474 this, url.c_str(), status.
ToString().c_str() );
476 log->
Error(
ZipMsg,
"[%p] Failed to open a ZIP archive (%s): %s",
477 this, url.c_str(), status.
ToString().c_str() );
482 Async( std::move( open_archive ), timeout );
494 if( !openfn.empty() || openstage != Done || !archive.
IsOpen() )
498 auto itr = cdmap.find( fn );
499 if( itr == cdmap.end() )
506 lfh.reset(
new LFH( fn, crc32, size, time( 0 ) ) );
507 log->
Dump(
ZipMsg,
"[%p] File %s opened for append.",
511 log->
Dump(
ZipMsg,
"[%p] Open failed: %s not in the ZIP archive.",
520 log->
Dump(
ZipMsg,
"[%p] Open failed: file exists %s, cannot append.",
527 log->
Dump(
ZipMsg,
"[%p] File %s opened for reading.",
540 eocd.reset(
new EOCD( cdoff, cdvec.size(), cdsize ) );
541 size += eocd->eocdSize ;
542 size += eocd->cdSize;
544 std::unique_ptr<ZIP64_EOCDL> zip64eocdl;
547 zip64eocd.reset(
new ZIP64_EOCD( cdoff, cdvec.size(), cdsize ) );
548 size += zip64eocd->zip64EocdTotalSize;
549 zip64eocdl.reset(
new ZIP64_EOCDL( *eocd, *zip64eocd ) );
555 metadata.reserve( size );
558 zip64eocd->Serialize( metadata );
560 zip64eocdl->Serialize( metadata );
561 eocd->Serialize( metadata );
569 void ZipArchive::SetCD(
const buffer_t &buffer )
571 if( openstage != NotParsed )
return;
573 const char *buff = buffer.data();
574 size_t size = buffer.size();
577 std::tie(cdvec, cdmap ) =
CDFH::Parse( buff, size );
579 orgcdsz = buff - buffer.data();
580 orgcdcnt = cdvec.size();
581 orgcdbuf.reserve( orgcdsz );
582 std::copy( buffer.data(), buff, std::back_inserter( orgcdbuf ) );
584 uint32_t signature = to<uint32_t>( buff );
588 buff += zip64eocd->zip64EocdTotalSize;
590 signature = to<uint32_t>( buff );
595 eocd.reset(
new EOCD( buff ) );
597 openstage = XrdCl::ZipArchive::Done;
616 std::vector<std::shared_ptr<buffer_t>> wrtbufs;
617 for(
auto &p : newfiles )
619 NewFile &nf = p.second;
620 if( !nf.overwrt )
continue;
621 uint32_t lfhlen = lfh->lfhSize;
622 auto lfhbuf = std::make_shared<buffer_t>();
623 lfhbuf->reserve( lfhlen );
624 nf.lfh->Serialize( *lfhbuf );
625 chunks.emplace_back( nf.offset, lfhbuf->size(), lfhbuf->data() );
626 wrtbufs.emplace_back( std::move( lfhbuf ) );
629 auto wrtbuff = std::make_shared<buffer_t>( GetCD() );
633 wrtbufs.emplace_back( std::move( wrtbuff ) );
635 std::vector<ChunkList> listsvec;
638 for(
auto itr = listsvec.rbegin(); itr != listsvec.rend(); ++itr)
644 p |=
Close( archive ) >>
647 if( st.
IsOK() ) Clear();
648 else openstage = Error;
653 log->
Dump(
ZipMsg,
"[%p] Successfully closed ZIP archive "
654 "(CD written).",
this );
656 log->
Error(
ZipMsg,
"[%p] Failed to close ZIP archive: %s",
659 if( handler ) handler->
HandleResponse( make_status( st ),
nullptr );
662 Async( std::move( p ), timeout );
675 log->
Dump(
ZipMsg,
"[%p] Successfully closed "
676 "ZIP archive.",
this );
681 log->
Error(
ZipMsg,
"[%p] Failed to close ZIP archive:"
682 " %s",
this, st.
ToString().c_str() );
687 Async( std::move( p ), timeout );
701 return ReadFromImpl<ChunkInfo>( *
this, fn, offset, size, buffer, handler, timeout );
714 return ReadFromImpl<PageInfo>( *
this, fn, offset, size, buffer, handler, timeout );
722 if( openstage != Done )
724 0,
"Archive not opened." );
732 std::unique_ptr<StatInfo> info( infoptr );
737 auto itr = cdvec.begin();
738 for( ; itr != cdvec.end() ; ++itr )
740 CDFH *cdfh = itr->get();
742 if( uncompressedSize == std::numeric_limits<uint32_t>::max() && cdfh->
extra )
743 uncompressedSize = cdfh->
extra->uncompressedSize;
744 StatInfo *entry_info = make_stat( *info, uncompressedSize );
762 std::vector<iovec>
iov( 2 );
768 std::shared_ptr<buffer_t> lfhbuf;
771 uint32_t lfhlen = lfh->lfhSize;
772 lfhbuf = std::make_shared<buffer_t>();
773 lfhbuf->reserve( lfhlen );
774 lfh->Serialize( *lfhbuf );
775 iov[0].iov_base = lfhbuf->
data();
776 iov[0].iov_len = lfhlen;
777 log->
Dump(
ZipMsg,
"[%p] Will write LFH.",
this );
784 iov[0].iov_base =
nullptr;
790 iov[1].iov_base =
const_cast<void*
>( buffer );
791 iov[1].iov_len = size;
793 uint64_t wrtoff = cdoff;
794 uint32_t wrtlen =
iov[0].iov_len +
iov[1].iov_len;
797 auto wrthandler = [=](
const XRootDStatus &st )
mutable
799 if( st.
IsOK() ) updated =
true;
809 if( archsize > cdoff )
819 if( archsize > cdoff && !ckpinit )
833 mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH;
834 cdvec.emplace_back(
new CDFH( lfh.get(), mode, wrtoff ) );
835 cdmap[openfn] = cdvec.size() - 1;
837 newfiles.emplace( std::piecewise_construct,
838 std::forward_as_tuple( lfh->filename ),
839 std::forward_as_tuple( wrtoff, std::move( lfh ) )
842 Async( std::move( p ), timeout );
843 return XRootDStatus();
851 if( openstage != Done || openfn.empty() )
857 auto itr = cdmap.find( openfn );
858 if( itr == cdmap.end() )
860 cdvec[itr->second]->ZCRC32 = crc32;
866 auto itr2 = newfiles.find( openfn );
867 if( itr2 == newfiles.end() )
869 itr2->second.lfh->ZCRC32 = crc32;
885 auto itr = cdmap.find( fn );
887 if( itr != cdmap.end() )
889 log->
Dump(
ZipMsg,
"[%p] Open failed: file exists %s, cannot append.",
894 log->
Dump(
ZipMsg,
"[%p] Appending file: %s.",
this, fn.c_str() );
898 lfh.reset(
new LFH( fn, crc32, size, time( 0 ) ) );
902 return WriteImpl( size, buffer, handler, timeout );
void Set(Type object, bool own=true)
static Log * GetLog()
Get default log.
void Add(ListEntry *entry)
Add an entry to the list - takes ownership.
void SetParentName(const std::string &parent)
Set name of the parent directory.
bool IsOpen() const
Check if the file is open.
bool GetProperty(const std::string &name, std::string &value) const
XRootDStatus Stat(bool force, ResponseHandler *handler, uint16_t timeout=0) XRD_WARN_UNUSED_RESULT
void Error(uint64_t topic, const char *format,...)
Report an error.
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
static void Repeat()
Repeat current operation.
static void Stop(const XRootDStatus &status=XrdCl::XRootDStatus())
Handle an async response.
virtual void HandleResponse(XRootDStatus *status, AnyObject *response)
std::string GetHostId() const
Get the host part of the URL (user:password@host:port)
const std::string & GetPath() const
Get the path.
static void SplitChunks(std::vector< ChunkList > &listsvec, const ChunkList &chunks, const uint32_t maxcs, const size_t maxc)
Split chunks in a ChunkList into one or more ChunkLists.
XRootDStatus ReadFrom(const std::string &fn, uint64_t offset, uint32_t size, void *buffer, ResponseHandler *handler, uint16_t timeout=0)
XRootDStatus UpdateMetadata(uint32_t crc32)
XRootDStatus OpenArchive(const std::string &url, OpenFlags::Flags flags, ResponseHandler *handler, uint16_t timeout=0)
XRootDStatus List(DirectoryList *&list)
XRootDStatus AppendFile(const std::string &fn, uint32_t crc32, uint32_t size, const void *buffer, ResponseHandler *handler, uint16_t timeout=0)
virtual ~ZipArchive()
Destructor.
XRootDStatus CloseArchive(ResponseHandler *handler, uint16_t timeout=0)
Create the central directory at the end of ZIP archive and close it.
XRootDStatus GetOffset(const std::string &fn, uint64_t &offset)
XRootDStatus PgReadFrom(const std::string &fn, uint64_t offset, uint32_t size, void *buffer, ResponseHandler *handler, uint16_t timeout=0)
ZipArchive(bool enablePlugIns=true)
Constructor.
XRootDStatus OpenFile(const std::string &fn, OpenFlags::Flags flags=OpenFlags::None, uint64_t size=0, uint32_t crc32=0)
Utility class for inflating a compressed buffer.
void QueueRsp(const XRootDStatus &st, uint64_t offset, buffer_t &&buffer)
void QueueReq(uint64_t offset, uint32_t length, void *buffer, ResponseHandler *handler)
VectorWriteImpl< false > VectorWrite(Ctx< File > file, Arg< ChunkList > chunks, uint16_t timeout=0)
Factory for creating VectorWriteImpl objects.
CheckpointImpl< false > Checkpoint(Ctx< File > file, Arg< ChkPtCode > code, uint16_t timeout=0)
Factory for creating ReadImpl objects.
WriteImpl< false > Write(Ctx< File > file, Arg< uint64_t > offset, Arg< uint32_t > size, Arg< const void * > buffer, uint16_t timeout=0)
Factory for creating WriteImpl objects.
const uint16_t stError
An error occurred that could potentially be retried.
std::future< XRootDStatus > Async(Pipeline pipeline, uint16_t timeout=0)
ReadImpl< false > Read(Ctx< File > file, Arg< uint64_t > offset, Arg< uint32_t > size, Arg< void * > buffer, uint16_t timeout=0)
Factory for creating ReadImpl objects.
const uint16_t errNotFound
ChkptWrtVImpl< false > ChkptWrtV(Ctx< File > file, Arg< uint64_t > offset, Arg< std::vector< iovec >> iov, uint16_t timeout=0)
Factory for creating ChkptWrtVImpl objects.
const uint16_t errDataError
data is corrupted
OpenImpl< false > Open(Ctx< File > file, Arg< std::string > url, Arg< OpenFlags::Flags > flags, Arg< Access::Mode > mode=Access::None, uint16_t timeout=0)
Factory for creating ReadImpl objects.
const uint16_t errInvalidOp
std::vector< ChunkInfo > ChunkList
List of chunks.
const uint16_t errNotSupported
WriteVImpl< false > WriteV(Ctx< File > file, Arg< uint64_t > offset, Arg< std::vector< iovec >> iov, uint16_t timeout=0)
Factory for creating WriteVImpl objects.
XRootDStatus ReadFromImpl(ZipArchive &me, const std::string &fn, uint64_t relativeOffset, uint32_t size, void *usrbuff, ResponseHandler *usrHandler, uint16_t timeout)
CloseImpl< false > Close(Ctx< File > file, uint16_t timeout=0)
Factory for creating CloseImpl objects.
none object for initializing empty Optional
std::vector< char > buffer_t
Describe a data chunk for vector read.
Flags
Open flags, may be or'd when appropriate.
@ Read
Open only for reading.
@ Update
Open for reading and writing.
bool IsOK() const
We're fine.
std::string ToString() const
Create a string representation.
uint32_t uncompressedSize
static uint64_t GetOffset(const CDFH &cdfh)
std::unique_ptr< Extra > extra
static std::tuple< cdvec_t, cdmap_t > Parse(const char *buffer, uint32_t bufferSize, uint16_t nbCdRecords)
uint16_t compressionMethod
static size_t CalcSize(const cdvec_t &cdvec, uint32_t orgcdsz, uint32_t orgcdcnt)
static void Serialize(uint32_t orgcdcnt, const buffer_t &orgcdbuf, const cdvec_t &cdvec, buffer_t &buffer)
static uint8_t GetSize(bool zip64)
static const uint16_t eocdBaseSize
static const uint16_t maxCommentLength
static const char * Find(const char *buffer, uint64_t size)
A data structure representing ZIP Local File Header.
A data structure representing the ZIP64 end of central directory locator.
static const uint32_t zip64EocdlSign
static const uint16_t zip64EocdlSize
static const uint32_t zip64EocdSign