-
Notifications
You must be signed in to change notification settings - Fork 149
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Compilable first draft of Repair Tool
Almost working except for the repair tool being destroyed due to a shared pointer This version works quite well, but uses mtfile Working version of the repair tool handles data and metadata corruption rebased to master Cleaned code rebased to master Last fixes rebased to master Restructured and commented repair tool functionality, the opening of archives can either be followed by a host replacement or on its own to detect and report on damaged archives to the user Added callback for just checking a file's checksum, but has a bug with checksum comparison in reader Fixed the archive opening and closing and the only instance that checks metadata is now the repair tool Archive closing waits until everything is written using a pipeline final and condition locking Fixed: We wait until all writes are finished before closing archives, WriteIntoFile is now a zip operation Fix: Wait for all replacement archives to be opened Added a test that corrupts two archives (or one archives in two spots), ran it successfully a couple hundred times Fixes: Corrupt archives at correct offsets and replaced two usages of uncompressedSize by compressedSize Fixed the actual root of the issue because corrupted offsets can lead to signflips in minus operations Small concurrency fix and formatting Concurrent version of repair Fixes to make more than nbdata+nbparity hosts possible, changed unit test to use 2 hosts more Cleaned some code Further cleanup Removed more unnecessary code and commented out lines Added a test to confirm that repair fails when too much is corrupted Added timeouts to the repair tool and a check in the opening process that checks whether one of the archives was blocked by the user. If it is the case, the tool exits immediately Adding some timeouts seems to introduce a bug, might have to revert? Trying to fix segfault for missing host Fixed an error that the open pipeline would fail if not all readHealth are successful Fixed the bugs, forgot a handler callback Changed atomics to relaxed memory order and implemented a block pool to limit the memory consumption Corrected a rebase mistake Removed file repair functionality to split it into a second PR
- Loading branch information
1 parent
e803d26
commit 12551ee
Showing
19 changed files
with
2,094 additions
and
513 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| #ifndef SRC_XRDEC_XRDECBLKPOOL_HH_ | ||
| #define SRC_XRDEC_XRDECBLKPOOL_HH_ | ||
|
|
||
| #include "XrdEc/XrdEcUtilities.hh" | ||
| #include "XrdEc/XrdEcObjCfg.hh" | ||
| #include "XrdEc/XrdEcConfig.hh" | ||
| #include "XrdEc/XrdEcThreadPool.hh" | ||
| #include "XrdEc/XrdEcReader.hh" | ||
|
|
||
| #include <vector> | ||
| #include <condition_variable> | ||
| #include <mutex> | ||
| #include <future> | ||
|
|
||
| namespace XrdEc | ||
| { | ||
| class BlockPool | ||
| { | ||
| public: | ||
|
|
||
| static BlockPool& Instance() | ||
| { | ||
| static BlockPool instance; | ||
| return instance; | ||
| } | ||
|
|
||
| std::shared_ptr<block_t> Create(ObjCfg &objcfg, Reader &reader, size_t blkid) | ||
| { | ||
| std::unique_lock<std::mutex> lck(mtx); | ||
| //--------------------------------------------------------------------- | ||
| // If pool is not empty, recycle existing buffer | ||
| //--------------------------------------------------------------------- | ||
| if (!pool.empty()) | ||
| { | ||
| std::shared_ptr<block_t> block(std::move(pool.front())); | ||
| pool.pop(); | ||
|
|
||
| // almost what the block constructor does except we dont make new stripes. | ||
| //block->reader = reader; | ||
| block->blkid = blkid; | ||
| block->state = std::vector<block_t::state_t>(objcfg.nbchunks, block_t::Empty); | ||
| block->pending = std::vector<block_t::pending_t>(objcfg.nbchunks); | ||
| block->recovering = 0; | ||
| block->redirectionIndex = 0; | ||
|
|
||
| return std::move(block); | ||
| } | ||
| //--------------------------------------------------------------------- | ||
| // Check if we can create a new buffer object without exceeding the | ||
| // the maximum size of the pool | ||
| //--------------------------------------------------------------------- | ||
| if (currentsize < totalsize) | ||
| { | ||
| std::shared_ptr<block_t> block = std::make_shared<block_t>(blkid, reader, objcfg); | ||
| ++currentsize; | ||
| return std::move(block); | ||
| } | ||
| //--------------------------------------------------------------------- | ||
| // If not, we have to wait until there is a buffer we can recycle | ||
| //--------------------------------------------------------------------- | ||
| while (pool.empty()) | ||
| cv.wait(lck); | ||
| std::shared_ptr<block_t> block(std::move(pool.front())); | ||
| pool.pop(); | ||
|
|
||
| // almost what the block constructor does except we dont make new stripes. | ||
| //block->reader = reader; | ||
| block->blkid = blkid; | ||
| block->state = std::vector<block_t::state_t>(objcfg.nbchunks, | ||
| block_t::Empty); | ||
| block->pending = std::vector<block_t::pending_t>(objcfg.nbchunks); | ||
| block->recovering = 0; | ||
| block->redirectionIndex = 0; | ||
|
|
||
| return std::move(block); | ||
| } | ||
|
|
||
| //----------------------------------------------------------------------- | ||
| //! Give back a buffer to the poool | ||
| //----------------------------------------------------------------------- | ||
| void Recycle(std::shared_ptr<block_t> block) | ||
| { | ||
| //if (block invalid) | ||
| // return; | ||
| std::unique_lock<std::mutex> lck(mtx); | ||
|
|
||
| pool.emplace(std::move(block)); | ||
| cv.notify_all(); | ||
| } | ||
|
|
||
| private: | ||
|
|
||
| //----------------------------------------------------------------------- | ||
| // Default constructor | ||
| //----------------------------------------------------------------------- | ||
| BlockPool() : | ||
| totalsize(1024), currentsize(0) | ||
| { | ||
| } | ||
|
|
||
| BlockPool(const BlockPool&) = delete; //< Copy constructor | ||
| BlockPool(BlockPool&&) = delete; //< Move constructor | ||
| BlockPool& operator=(const BlockPool&) = delete; //< Copy assigment operator | ||
| BlockPool& operator=(BlockPool&&) = delete; //< Move assigment operator | ||
|
|
||
| const size_t totalsize; //< maximum size of the pool | ||
| size_t currentsize; //< current size of the pool | ||
| std::condition_variable cv; | ||
| std::mutex mtx; | ||
| std::queue<std::shared_ptr<block_t>> pool; //< the pool itself | ||
| }; | ||
| } | ||
|
|
||
| #endif /* SRC_XRDEC_XRDECBLKPOOL_HH_ */ |
Oops, something went wrong.