/**************************************************************************/ /* */ /* Copyright (c) 2001, 2011 NoMachine (http://www.nomachine.com) */ /* Copyright (c) 2008-2014 Oleksandr Shneyder <o.shneyder@phoca-gmbh.de> */ /* Copyright (c) 2014-2016 Ulrich Sibiller <uli42@gmx.de> */ /* Copyright (c) 2014-2016 Mihai Moldovan <ionic@ionic.de> */ /* Copyright (c) 2011-2016 Mike Gabriel <mike.gabriel@das-netzwerkteam.de>*/ /* Copyright (c) 2015-2016 Qindel Group (http://www.qindel.com) */ /* */ /* NXCOMP, NX protocol compression and NX extensions to this software */ /* are copyright of the aforementioned persons and companies. */ /* */ /* Redistribution and use of the present software is allowed according */ /* to terms specified in the file LICENSE.nxcomp which comes in the */ /* source distribution. */ /* */ /* All rights reserved. */ /* */ /* NOTE: This software has received contributions from various other */ /* contributors, only the core maintainers and supporters are listed as */ /* copyright holders. Please contact us, if you feel you should be listed */ /* as copyright holder, as well. */ /* */ /**************************************************************************/ #include <unistd.h> #include <cstring> #include <sys/stat.h> #include <sys/types.h> #include <utime.h> #include "Misc.h" #include "Split.h" #include "Control.h" #include "Statistics.h" #include "EncodeBuffer.h" #include "DecodeBuffer.h" #include "StaticCompressor.h" #include "Unpack.h" // // Set the verbosity level. // #define PANIC #define WARNING #undef TEST #undef DEBUG #undef DUMP // // Define this to trace elements // allocated and deallocated. // #undef REFERENCES // // Counters used for store control. // int SplitStore::totalSplitSize_; int SplitStore::totalSplitStorageSize_; // // This is used for reference count. // #ifdef REFERENCES int Split::references_ = 0; #endif Split::Split() { resource_ = nothing; position_ = nothing; store_ = NULL; d_size_ = 0; i_size_ = 0; c_size_ = 0; r_size_ = 0; next_ = 0; load_ = 0; save_ = 0; checksum_ = NULL; state_ = split_undefined; mode_ = split_none; action_ = is_discarded; #ifdef REFERENCES references_++; *logofs << "Split: Created new Split at " << this << " out of " << references_ << " allocated references.\n" << logofs_flush; #endif } Split::~Split() { delete [] checksum_; #ifdef REFERENCES references_--; *logofs << "Split: Deleted Split at " << this << " out of " << references_ << " allocated references.\n" << logofs_flush; #endif } SplitStore::SplitStore(StaticCompressor *compressor, CommitStore *commits, int resource) : compressor_(compressor), commits_(commits), resource_(resource) { splits_ = new T_splits(); current_ = splits_ -> end(); splitStorageSize_ = 0; #ifdef TEST *logofs << "SplitStore: Created new store ["; if (resource_ != nothing) { *logofs << resource_; } else { *logofs << "commit"; } *logofs << "].\n" << logofs_flush; *logofs << "SplitStore: Total messages in stores are " << totalSplitSize_ << " with total storage size " << totalSplitStorageSize_ << ".\n" << logofs_flush; #endif } SplitStore::~SplitStore() { totalSplitSize_ -= splits_ -> size(); totalSplitStorageSize_ -= splitStorageSize_; for (T_splits::iterator i = splits_ -> begin(); i != splits_ -> end(); i++) { delete *i; } delete splits_; #ifdef TEST *logofs << "SplitStore: Deleted store ["; if (resource_ != nothing) { *logofs << resource_; } else { *logofs << "commit"; } *logofs << "] with storage size " << splitStorageSize_ << ".\n" << logofs_flush; *logofs << "SplitStore: Total messages in stores are " << totalSplitSize_ << " with total storage size " << totalSplitStorageSize_ << ".\n" << logofs_flush; #endif } // // This is called at the encoding side. // Split *SplitStore::add(MessageStore *store, int resource, T_split_mode mode, int position, T_store_action action, T_checksum checksum, const unsigned char *buffer, const int size) { #ifdef TEST *logofs << "SplitStore: Adding message [" << (unsigned int) store -> opcode() << "] resource " << resource << " mode " << mode << " position " << position << " action [" << DumpAction(action) << "] and checksum [" << DumpChecksum(checksum) << "]" << ".\n" << logofs_flush; #endif Split *split = new Split(); if (split == NULL) { #ifdef PANIC *logofs << "SplitStore: PANIC! Can't allocate " << "memory for the split.\n" << logofs_flush; #endif cerr << "Error" << ": Can't allocate memory " << "for the split.\n"; HandleAbort(); } split -> store_ = store; split -> resource_ = resource; split -> mode_ = mode; split -> position_ = position; split -> action_ = action; split -> store_ -> validateSize(size); // // The checksum is not provided if the // message is cached. // if (checksum != NULL) { split -> checksum_ = new md5_byte_t[MD5_LENGTH]; memcpy(split -> checksum_, checksum, MD5_LENGTH); } // // We don't need the identity data at the // encoding side. This qualifies the split // as a split generated at the encoding // side. // split -> i_size_ = store -> identitySize(buffer, size); split -> d_size_ = size - split -> i_size_; if (action == IS_ADDED || action == is_discarded) { // // If the message was added to message // store or discarded we need to save // the real data so we can transfer it // at later time. // split -> data_.resize(split -> d_size_); memcpy(split -> data_.begin(), buffer + split -> i_size_, split -> d_size_); // // If the message was added, lock it so // it will not be used by the encoding // side until it is recomposed. // if (action == IS_ADDED) { split -> store_ -> lock(split -> position_); #ifdef TEST commits_ -> validate(split); #endif } } #ifdef WARNING else { *logofs << "SplitStore: WARNING! Not copying data for the cached message.\n" << logofs_flush; } #endif push(split); return split; } // // This is called at decoding side. If checksum // is provided, the message can be searched on // disk, then, if message is found, an event is // sent to abort the data transfer. // Split *SplitStore::add(MessageStore *store, int resource, int position, T_store_action action, T_checksum checksum, unsigned char *buffer, const int size) { #ifdef TEST *logofs << "SplitStore: Adding message [" << (unsigned int) store -> opcode() << "] resource " << resource << " position " << position << " action [" << DumpAction(action) << "] and checksum [" << DumpChecksum(checksum) << "].\n" << logofs_flush; #endif Split *split = new Split(); if (split == NULL) { #ifdef PANIC *logofs << "SplitStore: PANIC! Can't allocate " << "memory for the split.\n" << logofs_flush; #endif cerr << "Error" << ": Can't allocate memory " << "for the split.\n"; HandleAbort(); } split -> store_ = store; split -> resource_ = resource; split -> position_ = position; split -> action_ = action; split -> store_ -> validateSize(size); // // Check if the checksum was provided // by the remote. // if (checksum != NULL) { split -> checksum_ = new md5_byte_t[MD5_LENGTH]; memcpy(split -> checksum_, checksum, MD5_LENGTH); } split -> i_size_ = store -> identitySize(buffer, size); // // Copy the identity so we can expand the // message when it is committed. // split -> identity_.resize(split -> i_size_); memcpy(split -> identity_.begin(), buffer, split -> i_size_); split -> d_size_ = size - split -> i_size_; if (action == IS_ADDED || action == is_discarded) { // // The unpack procedure will check if the // first 2 bytes of the buffer contain the // pattern and will not try to expand the // image. // split -> data_.resize(2); unsigned char *data = split -> data_.begin(); data[0] = SPLIT_PATTERN; data[1] = SPLIT_PATTERN; // // If the message was added to the store, // we don't have the data part, yet, so // we need to lock the message until it // is recomposed. // if (action == IS_ADDED) { split -> store_ -> lock(split -> position_); #ifdef TEST commits_ -> validate(split); #endif } } else { #ifdef WARNING *logofs << "SplitStore: WARNING! Copying data for the cached message.\n" << logofs_flush; #endif // // We may optionally take the data from the // message store in compressed form, but, // as the data has been decompressed in the // buffer, we save a further decompression. // split -> data_.resize(split -> d_size_); memcpy(split -> data_.begin(), buffer + split -> i_size_, split -> d_size_); } push(split); return split; } void SplitStore::push(Split *split) { splits_ -> push_back(split); splitStorageSize_ += getNodeSize(split); totalSplitSize_++; totalSplitStorageSize_ += getNodeSize(split); statistics -> addSplit(); #ifdef TEST *logofs << "SplitStore: There are " << splits_ -> size() << " messages in store [" << resource_ << "] with " << "storage size " << splitStorageSize_ << ".\n" << logofs_flush; *logofs << "SplitStore: Total messages in stores are " << totalSplitSize_ << " with total storage size " << totalSplitStorageSize_ << ".\n" << logofs_flush; #endif split -> state_ = split_added; } void SplitStore::dump() { #ifdef DUMP int n; Split *split; *logofs << "SplitStore: DUMP! Dumping content of "; if (commits_ == NULL) { *logofs << "[commits]"; } else { *logofs << "[splits] for store [" << resource_ << "]"; } *logofs << " with [" << getSize() << "] elements " << "in the store.\n" << logofs_flush; n = 0; for (T_splits::iterator i = splits_ -> begin(); i != splits_ -> end(); i++, n++) { split = *i; *logofs << "SplitStore: DUMP! Split [" << n << "] has action [" << DumpAction(split -> action_) << "] state [" << DumpState(split -> state_) << "] "; if (split -> resource_ >= 0) { *logofs << "resource " << split -> resource_; } *logofs << " request " << (unsigned) split -> store_ -> opcode() << " position " << split -> position_ << " size is " << split -> data_.size() << " (" << split -> d_size_ << "/" << split -> c_size_ << "/" << split -> r_size_ << ") with " << split -> data_.size() - split -> next_ << "] bytes to go.\n" << logofs_flush; } #endif } int SplitStore::send(EncodeBuffer &encodeBuffer, int packetSize) { if (splits_ -> size() == 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Function send called with no splits available.\n" << logofs_flush; #endif cerr << "Error" << ": Function send called with no splits available.\n"; HandleAbort(); } // // A start operation must always be executed on // the split, even in the case the split will be // later aborted. // if (current_ == splits_ -> end()) { start(encodeBuffer); } // // If we have matched the checksum received from // the remote side then we must abort the current // split, else we can send another block of data // to the remote peer. // Split *split = *current_; unsigned int abort = 0; if (split -> state_ == split_loaded) { abort = 1; } encodeBuffer.encodeBoolValue(abort); if (abort == 1) { #ifdef TEST *logofs << "SplitStore: Aborting split for checksum [" << DumpChecksum(split -> checksum_) << "] position " << split -> position_ << " with " << (split -> data_.size() - split -> next_) << " bytes to go " << "out of " << split -> data_.size() << ".\n" << logofs_flush; #endif statistics -> addSplitAborted(); statistics -> addSplitAbortedBytesOut(split -> data_.size() - split -> next_); split -> next_ = split -> data_.size(); split -> state_ = split_aborted; } else { int count = (packetSize <= 0 || split -> next_ + packetSize > (int) split -> data_.size() ? split -> data_.size() - split -> next_ : packetSize); #ifdef TEST *logofs << "SplitStore: Sending split for checksum [" << DumpChecksum(split -> checksum_) << "] count " << count << " position " << split -> position_ << ". Data size is " << split -> data_.size() << " (" << split -> d_size_ << "/" << split -> c_size_ << "), " << split -> data_.size() - (split -> next_ + count) << " to go.\n" << logofs_flush; #endif encodeBuffer.encodeValue(count, 32, 10); encodeBuffer.encodeMemory(split -> data_.begin() + split -> next_, count); split -> next_ += count; } // // Was data completely transferred? We are the // sending side. We must update the message in // store, even if split was aborted. // if (split -> next_ != ((int) split -> data_.size())) { return 0; } // // Move the split at the head of the // list to the commits. // remove(split); // // Reset current position to the // end of repository. // current_ = splits_ -> end(); #ifdef TEST *logofs << "SplitStore: Removed split at head of the list. " << "Resource is " << split -> resource_ << " request " << (unsigned) split -> store_ -> opcode() << " position " << split -> position_ << ".\n" << logofs_flush; #endif return 1; } int SplitStore::start(EncodeBuffer &encodeBuffer) { // // Get the element at the top of the // list. // current_ = splits_ -> begin(); Split *split = *current_; #ifdef TEST *logofs << "SplitStore: Starting split for checksum [" << DumpChecksum(split -> checksum_) << "] position " << split -> position_ << " with " << (split -> data_.size() - split -> next_) << " bytes to go " << "out of " << split -> data_.size() << ".\n" << logofs_flush; #endif // // See if compression of the data part is // enabled. // if (split -> store_ -> enableCompress) { // // If the split is going to be aborted don't // compress the data and go straight to the // send. The new data size will be assumed // from the disk cache. // if (split -> state_ != split_loaded) { unsigned int compressedSize = 0; unsigned char *compressedData = NULL; if (control -> LocalDataCompression && (compressor_ -> compressBuffer(split -> data_.begin(), split -> d_size_, compressedData, compressedSize))) { // // Replace the data with the one in // compressed form. // #ifdef TEST *logofs << "SplitStore: Split data of size " << split -> d_size_ << " has been compressed to " << compressedSize << " bytes.\n" << logofs_flush; #endif split -> data_.clear(); split -> data_.resize(compressedSize); memcpy(split -> data_.begin(), compressedData, compressedSize); split -> c_size_ = compressedSize; // // Inform our peer that the data is // compressed and send the new size. // encodeBuffer.encodeBoolValue(1); encodeBuffer.encodeValue(compressedSize, 32, 14); #ifdef TEST *logofs << "SplitStore: Signaled " << split -> c_size_ << " bytes of compressed data for this message.\n" << logofs_flush; #endif return 1; } } #ifdef TEST else { *logofs << "SplitStore: Not trying to compress the " << "loaded message.\n" << logofs_flush; } #endif // // Tell to the remote that data will // follow uncompressed. // encodeBuffer.encodeBoolValue(0); } return 1; } int SplitStore::start(DecodeBuffer &decodeBuffer) { #ifdef TEST *logofs << "SplitStore: Going to receive a new split from the remote side.\n" << logofs_flush; #endif // // Get the element at the head // of the list. // current_ = splits_ -> begin(); Split *split = *current_; unsigned int compressedSize = 0; // // Save the data size known by the remote. // This information will be needed if the // remote will not have a chance to abort // the split. // split -> r_size_ = split -> d_size_; // // Find out if data was compressed by the // remote. // if (split -> store_ -> enableCompress) { decodeBuffer.decodeBoolValue(compressedSize); if (compressedSize == 1) { // // Get the compressed size. // // Since ProtoStep7 (#issue 108) decodeBuffer.decodeValue(compressedSize, 32, 14); split -> store_ -> validateSize(split -> d_size_, compressedSize); split -> r_size_ = compressedSize; } } // // Update the size if the split // was not already loaded. // if (split -> state_ != split_loaded) { split -> data_.clear(); if (compressedSize > 0) { split -> c_size_ = compressedSize; #ifdef TEST *logofs << "SplitStore: Split data of size " << split -> d_size_ << " was compressed to " << split -> c_size_ << " bytes.\n" << logofs_flush; #endif split -> data_.resize(split -> c_size_); } else { split -> data_.resize(split -> d_size_); } unsigned char *data = split -> data_.begin(); data[0] = SPLIT_PATTERN; data[1] = SPLIT_PATTERN; } #ifdef TEST else { // // The message had been already // loaded from disk. // if (compressedSize > 0) { if ((int) compressedSize != split -> c_size_) { *logofs << "SplitStore: WARNING! Compressed data size is " << "different than the loaded compressed size.\n" << logofs_flush; } *logofs << "SplitStore: Ignoring the new size with " << "loaded compressed size " << split -> c_size_ << ".\n" << logofs_flush; } } #endif return 1; } int SplitStore::receive(DecodeBuffer &decodeBuffer) { if (splits_ -> size() == 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Function receive called with no splits available.\n" << logofs_flush; #endif cerr << "Error" << ": Function receive called with no splits available.\n"; HandleAbort(); } if (current_ == splits_ -> end()) { start(decodeBuffer); } // // Check first if split was aborted, else add // any new data to message being recomposed. // Split *split = *current_; unsigned int abort = 0; decodeBuffer.decodeBoolValue(abort); if (abort == 1) { #ifdef TEST *logofs << "SplitStore: Aborting split for checksum [" << DumpChecksum(split -> checksum_) << "] position " << split -> position_ << " with " << (split -> data_.size() - split -> next_) << " bytes to go " << "out of " << split -> data_.size() << ".\n" << logofs_flush; #endif statistics -> addSplitAborted(); statistics -> addSplitAbortedBytesOut(split -> r_size_ - split -> next_); split -> next_ = split -> r_size_; split -> state_ = split_aborted; } else { // // Get the size of the packet. // unsigned int count; decodeBuffer.decodeValue(count, 32, 10); // // If the split was not already loaded from // disk, decode the packet and update our // copy of the data. The encoding side may // have not received the abort event, yet, // and may be unaware that the message is // stored in compressed form at our side. // #ifdef TEST *logofs << "SplitStore: Receiving split for checksum [" << DumpChecksum(split -> checksum_) << "] count " << count << " position " << split -> position_ << ". Data size is " << split -> data_.size() << " (" << split -> d_size_ << "/" << split -> c_size_ << "/" << split -> r_size_ << "), " << split -> r_size_ - (split -> next_ + count) << " to go.\n" << logofs_flush; #endif if (split -> next_ + count > (unsigned) split -> r_size_) { #ifdef PANIC *logofs << "SplitStore: PANIC! Invalid data count " << count << "provided in the split.\n" << logofs_flush; *logofs << "SplitStore: PANIC! While receiving split for " << "checksum [" << DumpChecksum(split -> checksum_) << "] with count " << count << " action [" << DumpAction(split -> action_) << "] state [" << DumpState(split -> state_) << "]. Data size is " << split -> data_.size() << " (" << split -> d_size_ << "/" << split -> c_size_ << "), " << split -> data_.size() - (split -> next_ + count) << " to go.\n" << logofs_flush; #endif cerr << "Error" << ": Invalid data count " << count << "provided in the split.\n"; HandleAbort(); } if (split -> state_ != split_loaded) { #ifdef TEST if (split -> next_ + count > split -> data_.size()) { #ifdef PANIC *logofs << "SplitStore: PANIC! Inconsistent split data size " << split -> data_.size() << " with expected size " << split -> r_size_ << ".\n" << logofs_flush; #endif HandleAbort(); } #endif memcpy(split -> data_.begin() + split -> next_, decodeBuffer.decodeMemory(count), count); } else { #ifdef TEST *logofs << "SplitStore: WARNING! Data discarded with split " << "loaded from disk.\n" << logofs_flush; #endif decodeBuffer.decodeMemory(count); } split -> next_ += count; } // // Is unsplit complete? // if (split -> next_ != split -> r_size_) { return 0; } // // If the persistent cache is enabled, // we have a valid checksum and the // split was not originally retrieved // from disk, save the message on disk. // if (split -> state_ != split_loaded && split -> state_ != split_aborted) { save(split); } // // Move the split at the head of the // list to the commits. // remove(split); // // Reset the current position to the // end of the repository. // current_ = splits_ -> end(); #ifdef TEST *logofs << "SplitStore: Removed split at head of the list. " << "Resource is " << split -> resource_ << " request " << (unsigned) split -> store_ -> opcode() << " position " << split -> position_ << ".\n" << logofs_flush; #endif return 1; } Split *SplitStore::pop() { if (splits_ -> size() == 0) { #ifdef TEST *logofs << "SplitStore: The split store is empty.\n" << logofs_flush; #endif return NULL; } // // Move the pointer at the end of the list. // The next send operation will eventually // start a new split. // current_ = splits_ -> end(); Split *split = *(splits_ -> begin()); splits_ -> pop_front(); #ifdef TEST *logofs << "SplitStore: Removed split at the head of the " << "list with resource " << split -> resource_ << " request " << (unsigned) split -> store_ -> opcode() << " position " << split -> position_ << ".\n" << logofs_flush; #endif splitStorageSize_ -= getNodeSize(split); totalSplitSize_--; totalSplitStorageSize_ -= getNodeSize(split); #ifdef TEST *logofs << "SplitStore: There are " << splits_ -> size() << " messages in store [" << resource_ << "] with " << "storage size " << splitStorageSize_ << ".\n" << logofs_flush; *logofs << "SplitStore: Total messages in stores are " << totalSplitSize_ << " with total storage size " << totalSplitStorageSize_ << ".\n" << logofs_flush; #endif return split; } void SplitStore::remove(Split *split) { #ifdef TEST *logofs << "SplitStore: Going to remove the split from the list.\n" << logofs_flush; #endif #ifdef TEST if (split != getFirstSplit()) { #ifdef PANIC *logofs << "SplitStore: PANIC! Trying to remove a split " << "not at the head of the list.\n" << logofs_flush; #endif cerr << "Error" << ": Trying to remove a split " << "not at the head of the list.\n"; HandleAbort(); } #endif // // Move the split to the commit store. // splits_ -> pop_front(); commits_ -> splits_ -> push_back(split); splitStorageSize_ -= getNodeSize(split); totalSplitSize_--; totalSplitStorageSize_ -= getNodeSize(split); #ifdef TEST *logofs << "SplitStore: There are " << splits_ -> size() << " messages in store [" << resource_ << "] with " << "storage size " << splitStorageSize_ << ".\n" << logofs_flush; *logofs << "SplitStore: Total messages in stores are " << totalSplitSize_ << " with total storage size " << totalSplitStorageSize_ << ".\n" << logofs_flush; #endif #ifdef TEST if (splits_ -> size() == 0) { if (splitStorageSize_ != 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Internal error calculating " << "split data size. It is " << splitStorageSize_ << " while should be 0.\n" << logofs_flush; #endif cerr << "Error" << ": Internal error calculating " << "split data size. It is " << splitStorageSize_ << " while should be 0.\n"; HandleAbort(); } } #endif } const char *SplitStore::name(const T_checksum checksum) { if (checksum == NULL) { return NULL; } char *pathName = control -> ImageCachePath; if (pathName == NULL) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot determine directory of " << "NX image files.\n" << logofs_flush; #endif return NULL; } int pathSize = strlen(pathName); // // File name is "[path][/I-c/I-][checksum][\0]", // where c is the first hex digit of checksum. // int nameSize = pathSize + 7 + MD5_LENGTH * 2 + 1; char *fileName = new char[nameSize]; if (fileName == NULL) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot allocate space for " << "NX image file name.\n" << logofs_flush; #endif return NULL; } strcpy(fileName, pathName); sprintf(fileName + pathSize, "/I-%1X/I-", *((unsigned char *) checksum) >> 4); for (unsigned int i = 0; i < MD5_LENGTH; i++) { sprintf(fileName + pathSize + 7 + (i * 2), "%02X", ((unsigned char *) checksum)[i]); } return fileName; } int SplitStore::save(Split *split) { // // Check if saving the message on the // persistent cache is enabled. // if (split -> save_ == 0) { return 0; } T_checksum checksum = split -> checksum_; const char *fileName = name(checksum); if (fileName == NULL) { return 0; } unsigned int splitSize; ostream *fileStream = NULL; unsigned char *fileHeader = NULL; // // Get the other data from the split. // unsigned char opcode = split -> store_ -> opcode(); unsigned char *data = split -> data_.begin(); int dataSize = split -> d_size_; int compressedSize = split -> c_size_; #ifdef DEBUG *logofs << "SplitStore: Going to save split OPCODE#" << (unsigned int) opcode << " to file '" << fileName << "' with size " << dataSize << " and compressed size " << compressedSize << ".\n" << logofs_flush; #endif DisableSignals(); // // Change the mask to make the file only // readable by the user, then restore the // old mask. // mode_t fileMode; // // Check if the file already exists. We try to // load the message when the split is started // and save it only if it is not found. Still // the remote side may send the same image mul- // tiple time and we may not have the time to // notify the abort. // struct stat fileStat; if (stat(fileName, &fileStat) == 0) { #ifdef TEST *logofs << "SplitStore: Image file '" << fileName << "' already present on disk.\n" << logofs_flush; #endif goto SplitStoreSaveError; } fileMode = umask(0077); fileStream = new ofstream(fileName, ios::out | ios::binary); umask(fileMode); if (CheckData(fileStream) < 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot open file '" << fileName << "' for output.\n" << logofs_flush; #endif goto SplitStoreSaveError; } fileHeader = new unsigned char[SPLIT_HEADER_SIZE]; if (fileHeader == NULL) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot allocate space for " << "NX image header.\n" << logofs_flush; #endif goto SplitStoreSaveError; } // // Leave 3 bytes for future use. Please note // that, on some CPUs, we can't use PutULONG() // to write integers that are not aligned to // the word boundary. // *fileHeader = opcode; *(fileHeader + 1) = 0; *(fileHeader + 2) = 0; *(fileHeader + 3) = 0; PutULONG(dataSize, fileHeader + 4, false); PutULONG(compressedSize, fileHeader + 8, false); splitSize = (compressedSize > 0 ? compressedSize : dataSize); if (PutData(fileStream, fileHeader, SPLIT_HEADER_SIZE) < 0 || PutData(fileStream, data, splitSize) < 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot write to NX " << "image file '" << fileName << "'.\n" << logofs_flush; #endif goto SplitStoreSaveError; } // // Check if all the data was written on the // disk and, if not, remove the faulty copy. // FlushData(fileStream); if (CheckData(fileStream) < 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Failed to write NX " << "image file '" << fileName << "'.\n" << logofs_flush; #endif cerr << "Warning" << ": Failed to write NX " << "image file '" << fileName << "'.\n"; goto SplitStoreSaveError; } #ifdef TEST *logofs << "SplitStore: Saved split to file '" << fileName << "' with data size " << dataSize << " and " << "compressed data size " << compressedSize << ".\n" << logofs_flush; #endif delete fileStream; delete [] fileName; delete [] fileHeader; EnableSignals(); // // Update the timestamp as the operation // may have taken some time. // getNewTimestamp(); return 1; SplitStoreSaveError: delete fileStream; if (fileName != NULL) { unlink(fileName); } delete [] fileName; delete [] fileHeader; EnableSignals(); return -1; } int SplitStore::find(Split *split) { const char *fileName = name(split -> checksum_); if (fileName == NULL) { return 0; } #ifdef DEBUG *logofs << "SplitStore: Going to find split OPCODE#" << (unsigned) split -> store_ -> opcode() << " in file '" << fileName << "'.\n" << logofs_flush; #endif // // Check if the file exists and, at the // same time, update the modification // time to prevent its deletion. // if (utime(fileName, NULL) == 0) { #ifdef TEST *logofs << "SplitStore: Found split OPCODE#" << (unsigned) split -> store_ -> opcode() << " in file '" << fileName << "'.\n" << logofs_flush; #endif delete [] fileName; return 1; } #ifdef TEST *logofs << "SplitStore: WARNING! Can't find split " << "OPCODE#" << (unsigned) split -> store_ -> opcode() << " in file '" << fileName << "'.\n" << logofs_flush; #endif delete [] fileName; return 0; } int SplitStore::load(Split *split) { // // Check if loading the image is enabled. // if (split -> load_ == 0) { return 0; } const char *fileName = name(split -> checksum_); if (fileName == NULL) { return 0; } unsigned char fileOpcode; int fileSize; int fileCSize; istream *fileStream = NULL; unsigned char *fileHeader = NULL; DisableSignals(); #ifdef DEBUG *logofs << "SplitStore: Going to load split OPCODE#" << (unsigned int) split -> store_ -> opcode() << " from file '" << fileName << "'.\n" << logofs_flush; #endif fileStream = new ifstream(fileName, ios::in | ios::binary); if (CheckData(fileStream) < 0) { #ifdef TEST *logofs << "SplitStore: WARNING! Can't open image file '" << fileName << "' on disk.\n" << logofs_flush; #endif goto SplitStoreLoadError; } fileHeader = new unsigned char[SPLIT_HEADER_SIZE]; if (fileHeader == NULL) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot allocate space for " << "NX image header.\n" << logofs_flush; #endif cerr << "Error" << ": Cannot allocate space for " << "NX image header.\n"; goto SplitStoreLoadError; } if (GetData(fileStream, fileHeader, SPLIT_HEADER_SIZE) < 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot read header from " << "NX image file '" << fileName << "'.\n" << logofs_flush; #endif cerr << "Warning" << ": Cannot read header from " << "NX image file '" << fileName << "'.\n"; goto SplitStoreLoadError; } fileOpcode = *fileHeader; fileSize = GetULONG(fileHeader + 4, false); fileCSize = GetULONG(fileHeader + 8, false); // // Don't complain if we find that data was saved // in compressed form even if we were not aware // of the compressed data size. The remote side // compresses the data only at the time it starts // the transferral of the split. We replace our // copy of the data with whatever we find on the // disk. // if (fileOpcode != split -> store_ -> opcode() || fileSize != split -> d_size_ || fileSize > control -> MaximumRequestSize || fileCSize > control -> MaximumRequestSize) { #ifdef TEST *logofs << "SplitStore: PANIC! Corrupted image file '" << fileName << "'. Expected " << (unsigned int) split -> store_ -> opcode() << "/" << split -> d_size_ << "/" << split -> c_size_ << " found " << (unsigned int) fileOpcode << "/" << fileSize << "/" << fileCSize << ".\n" << logofs_flush; #endif cerr << "Warning" << ": Corrupted image file '" << fileName << "'. Expected " << (unsigned int) split -> store_ -> opcode() << "/" << split -> d_size_ << "/" << split -> c_size_ << " found " << (unsigned int) fileOpcode << "/" << fileSize << "/" << fileCSize << ".\n"; goto SplitStoreLoadError; } // // Update the data size with the size // we got from the disk record. // split -> d_size_ = fileSize; split -> c_size_ = fileCSize; unsigned int splitSize; if (fileCSize > 0) { splitSize = fileCSize; } else { splitSize = fileSize; } // // Allocate a new buffer if we didn't // do that already or if the size is // different. // if (split -> data_.size() != splitSize) { split -> data_.clear(); split -> data_.resize(splitSize); } if (GetData(fileStream, split -> data_.begin(), splitSize) < 0) { #ifdef PANIC *logofs << "SplitStore: PANIC! Cannot read data from " << "NX image file '" << fileName << "'.\n" << logofs_flush; #endif cerr << "Warning" << ": Cannot read data from " << "NX image file '" << fileName << "'.\n"; goto SplitStoreLoadError; } delete fileStream; delete [] fileHeader; delete [] fileName; EnableSignals(); // // Update the timestamp as the operation // may have taken some time. // getNewTimestamp(); return 1; SplitStoreLoadError: delete fileStream; unlink(fileName); delete [] fileName; delete [] fileHeader; EnableSignals(); return -1; } Split *CommitStore::pop() { if (splits_ -> size() == 0) { #ifdef TEST *logofs << "CommitStore: The commit store is empty.\n" << logofs_flush; #endif return NULL; } Split *split = *(splits_ -> begin()); splits_ -> pop_front(); #ifdef TEST *logofs << "CommitStore: Removed commit split at the head " << "of the list with resource " << split -> resource_ << " request " << (unsigned) split -> store_ -> opcode() << " position " << split -> position_ << ".\n" << logofs_flush; #endif return split; } int CommitStore::expand(Split *split, unsigned char *buffer, const int size) { #ifdef TEST *logofs << "CommitStore: Expanding split data with " << size << " bytes to write.\n" << logofs_flush; #endif #ifdef TEST if (size < split -> i_size_ + split -> d_size_) { #ifdef PANIC *logofs << "CommitStore: PANIC! Wrong size of the provided " << "buffer. It should be " << split -> i_size_ + split -> d_size_ << " instead of " << size << ".\n" << logofs_flush; #endif cerr << "Error" << ": Wrong size of the provided " << "buffer. It should be " << split -> i_size_ + split -> d_size_ << " instead of " << size << ".\n"; HandleAbort(); } #endif #ifdef DEBUG *logofs << "CommitStore: Copying " << split -> i_size_ << " bytes of identity.\n" << logofs_flush; #endif memcpy(buffer, split -> identity_.begin(), split -> i_size_); // // Copy data, if any, to the buffer. // if (size > split -> i_size_) { // // Check if message has been stored // in compressed format. // if (split -> c_size_ == 0) { #ifdef DEBUG *logofs << "CommitStore: Copying " << split -> d_size_ << " bytes of plain data.\n" << logofs_flush; #endif memcpy(buffer + split -> i_size_, split -> data_.begin(), split -> d_size_); } else { #ifdef DEBUG *logofs << "CommitStore: Decompressing " << split -> c_size_ << " bytes and copying " << split -> d_size_ << " bytes of data.\n" << logofs_flush; #endif if (compressor_ -> decompressBuffer(buffer + split -> i_size_, split -> d_size_, split -> data_.begin(), split -> c_size_) < 0) { #ifdef PANIC *logofs << "CommitStore: PANIC! Split data decompression failed.\n" << logofs_flush; #endif cerr << "Error" << ": Split data decompression failed.\n"; return -1; } } } return 1; } int CommitStore::update(Split *split) { if (split -> action_ != IS_ADDED) { return 0; } // // We don't need the identity data at // the encoding side. // if (split -> identity_.size() == 0) { #ifdef TEST *logofs << "SplitStore: Going to update the size " << "for object at position " << split -> position_ << " with data size " << split -> d_size_ << " and compressed data size " << split -> c_size_ << ".\n" << logofs_flush; #endif split -> store_ -> updateData(split -> position_, split -> d_size_, split -> c_size_); } else { #ifdef TEST *logofs << "SplitStore: Going to update data and size " << "for object at position " << split -> position_ << " with data size " << split -> d_size_ << " and compressed data size " << split -> c_size_ << ".\n" << logofs_flush; #endif split -> store_ -> updateData(split -> position_, split -> data_.begin(), split -> d_size_, split -> c_size_); } // // Unlock message so that we can remove // or save it on disk at shutdown. // if (split -> action_ == IS_ADDED) { split -> store_ -> unlock(split -> position_); #ifdef TEST validate(split); #endif } return 1; } int CommitStore::validate(Split *split) { MessageStore *store = split -> store_; int p, n, s; s = store -> cacheSlots; for (p = 0, n = 0; p < s; p++) { if (store -> getLocks(p) == 1) { n++; } else if (store -> getLocks(p) != 0) { #ifdef PANIC *logofs << "CommitStore: PANIC! Repository for OPCODE#" << (unsigned int) store -> opcode() << " has " << store -> getLocks(p) << " locks for message " << "at position " << p << ".\n" << logofs_flush; #endif cerr << "Error" << ": Repository for OPCODE#" << (unsigned int) store -> opcode() << " has " << store -> getLocks(p) << " locks for message " << "at position " << p << ".\n"; HandleAbort(); } } #ifdef TEST *logofs << "CommitStore: Repository for OPCODE#" << (unsigned int) store -> opcode() << " has " << n << " locked messages.\n" << logofs_flush; #endif return 1; }