Loading...
Searching...
No Matches
XrdEcRedundancyProvider.cc
Go to the documentation of this file.
/************************************************************************
* KineticIo - a file io interface library to kinetic devices. *
* *
* This Source Code Form is subject to the terms of the Mozilla *
* Public License, v. 2.0. If a copy of the MPL was not *
* distributed with this file, You can obtain one at *
* https://mozilla.org/MP:/2.0/. *
* *
* This program is distributed in the hope that it will be useful, *
* but is provided AS-IS, WITHOUT ANY WARRANTY; including without *
* the implied warranty of MERCHANTABILITY, NON-INFRINGEMENT or *
* FITNESS FOR A PARTICULAR PURPOSE. See the Mozilla Public *
* License for more details. *
************************************************************************/
#include "XrdEc/XrdEcRedundancyProvider.hh"
#include <isa-l.h>
#include <cstring>
#include <sstream>
#include <algorithm>
namespace XrdEc
{
//--------------------------------------------------------------------------
//--------------------------------------------------------------------------
class Convert{
public:
//--------------------------------------------------------------------------
//--------------------------------------------------------------------------
template<typename...Args>
static std::string toString(Args&&...args){
std::stringstream s;
argsToStream(s, std::forward<Args>(args)...);
return s.str();
}
private:
//--------------------------------------------------------------------------
//--------------------------------------------------------------------------
template<typename Last>
static void argsToStream(std::stringstream& stream, Last&& last) {
stream << last;
}
//--------------------------------------------------------------------------
//--------------------------------------------------------------------------
template<typename First, typename...Rest >
static void argsToStream(std::stringstream& stream, First&& first, Rest&&...rest) {
stream << first;
argsToStream(stream, std::forward<Rest>(rest)...);
}
};
/* This function is (almost) completely ripped from the erasure_code_test.cc file
distributed with the isa-l library. */
static int gf_gen_decode_matrix(
unsigned char* encode_matrix, // in: encode matrix
unsigned char* decode_matrix, // in: buffer, out: generated decode matrix
unsigned int* decode_index, // out: order of healthy blocks used for decoding [data#1, data#3, ..., parity#1... ]
unsigned char* src_err_list, // in: array of #nerrs size [index error #1, index error #2, ... ]
unsigned char* src_in_err, // in: array of #data size > [1,0,0,0,1,0...] -> 0 == no error, 1 == error
unsigned int nerrs, // #total errors
unsigned int nsrcerrs, // #data errors
unsigned int k, // #data
unsigned int m // #data+parity
)
{
unsigned i, j, p;
unsigned int r;
unsigned char* invert_matrix, * backup, * b, s;
int incr = 0;
size_t mk = (size_t)m * (size_t)k;
std::vector<unsigned char> memory(3 * mk);
b = &memory[0];
backup = &memory[mk];
invert_matrix = &memory[2 * mk];
// Construct matrix b by removing error rows
for (i = 0, r = 0; i < k; i++, r++) {
while (src_in_err[r]) {
r++;
}
for (j = 0; j < k; j++) {
b[k * i + j] = encode_matrix[k * r + j];
backup[k * i + j] = encode_matrix[k * r + j];
}
decode_index[i] = r;
}
incr = 0;
while (gf_invert_matrix(b, invert_matrix, k) < 0) {
if (nerrs == (m - k)) {
return -1;
}
incr++;
memcpy(b, backup, mk);
for (i = nsrcerrs; i < nerrs - nsrcerrs; i++) {
if (src_err_list[i] == (decode_index[k - 1] + incr)) {
// skip the erased parity line
incr++;
continue;
}
}
if (decode_index[k - 1] + incr >= m) {
return -1;
}
decode_index[k - 1] += incr;
for (j = 0; j < k; j++) {
b[k * (k - 1) + j] = encode_matrix[k * decode_index[k - 1] + j];
}
};
for (i = 0; i < nsrcerrs; i++) {
for (j = 0; j < k; j++) {
decode_matrix[k * i + j] = invert_matrix[k * src_err_list[i] + j];
}
}
/* src_err_list from encode_matrix * invert of b for parity decoding */
for (p = nsrcerrs; p < nerrs; p++) {
for (i = 0; i < k; i++) {
s = 0;
for (j = 0; j < k; j++) {
s ^= gf_mul(invert_matrix[j * k + i],
encode_matrix[k * src_err_list[p] + j]);
}
decode_matrix[k * p + i] = s;
}
}
return 0;
}
RedundancyProvider::RedundancyProvider( const ObjCfg &objcfg ) :
objcfg( objcfg ),
encode_matrix( objcfg.nbchunks * objcfg.nbdata )
{
// k = data
// m = data + parity
gf_gen_cauchy1_matrix( encode_matrix.data(), static_cast<int>( objcfg.nbchunks ), static_cast<int>( objcfg.nbdata ) );
}
std::string RedundancyProvider::getErrorPattern( stripes_t &stripes ) const
{
std::string pattern( objcfg.nbchunks, 0 );
for( uint8_t i = 0; i < objcfg.nbchunks; ++i )
if( !stripes[i].valid ) pattern[i] = '\1';
return pattern;
}
RedundancyProvider::CodingTable& RedundancyProvider::getCodingTable( const std::string& pattern )
{
std::lock_guard<std::mutex> lock(mutex);
/* If decode matrix is not already cached we have to construct it. */
if( !cache.count(pattern) )
{
/* Expand pattern */
int nerrs = 0, nsrcerrs = 0;
unsigned char err_indx_list[objcfg.nbparity];
/* Avoid narrowing cast warning, size is always < 256 */
uint8_t n = static_cast<uint8_t>(pattern.size() & 0xff);
for (uint8_t i = 0; i < n; i++) {
if (pattern[i]) {
if (nerrs == objcfg.nbparity)
throw IOError( XrdCl::XRootDStatus( XrdCl::stError, XrdCl::errDataError, errno, "More errors than parity blocks" ) );
err_indx_list[nerrs++] = i;
if (i < objcfg.nbdata) { nsrcerrs++; }
}
}
/* Allocate Decode Object. */
CodingTable dd;
dd.nErrors = nerrs;
dd.blockIndices.resize( objcfg.nbdata );
dd.table.resize( objcfg.nbdata * objcfg.nbparity * 32);
/* Compute decode matrix. */
std::vector<unsigned char> decode_matrix(objcfg.nbchunks * objcfg.nbdata);
if (gf_gen_decode_matrix( encode_matrix.data(), decode_matrix.data(), dd.blockIndices.data(),
err_indx_list, (unsigned char*) pattern.c_str(), nerrs, nsrcerrs,
static_cast<int>( objcfg.nbdata ), static_cast<int>( objcfg.nbchunks ) ) )
throw IOError( XrdCl::XRootDStatus( XrdCl::stError, XrdCl::errDataError, errno, "Failed computing decode matrix" ) );
/* Compute Tables. */
ec_init_tables( static_cast<int>( objcfg.nbdata ), nerrs, decode_matrix.data(), dd.table.data() );
cache.insert( std::make_pair(pattern, dd) );
}
return cache.at(pattern);
}
void RedundancyProvider::replication( stripes_t &stripes )
{
// get index of a valid block
void *healthy = nullptr;
for( auto itr = stripes.begin(); itr != stripes.end(); ++itr )
{
if( itr->valid )
healthy = itr->buffer;
}
if( !healthy ) throw IOError( XrdCl::XRootDStatus( XrdCl::stError, XrdCl::errDataError ) );
// now replicate, by now 'buffers' should contain all chunks
for( uint8_t i = 0; i < objcfg.nbchunks; ++i )
{
if( !stripes[i].valid )
memcpy( stripes[i].buffer, healthy, objcfg.chunksize );
}
}
void RedundancyProvider::compute( stripes_t &stripes )
{
/* throws if stripe is not recoverable */
std::string pattern = getErrorPattern( stripes );
/* nothing to do if there are no parity blocks. */
if ( !objcfg.nbparity ) return;
/* in case of a single data block use replication */
if ( objcfg.nbdata == 1 )
return replication( stripes );
/* normal operation: erasure coding */
CodingTable& dd = getCodingTable(pattern);
unsigned char* inbuf[objcfg.nbdata];
for( uint8_t i = 0; i < objcfg.nbdata; i++ )
inbuf[i] = reinterpret_cast<unsigned char*>( stripes[dd.blockIndices[i]].buffer );
std::vector<unsigned char> memory( dd.nErrors * objcfg.chunksize );
unsigned char* outbuf[dd.nErrors];
for (int i = 0; i < dd.nErrors; i++)
{
outbuf[i] = &memory[i * objcfg.chunksize];
}
ec_encode_data(
static_cast<int>( objcfg.chunksize ), // Length of each block of data (vector) of source or destination data.
static_cast<int>( objcfg.nbdata ), // The number of vector sources in the generator matrix for coding.
dd.nErrors, // The number of output vectors to concurrently encode/decode.
dd.table.data(), // Pointer to array of input tables
inbuf, // Array of pointers to source input buffers
outbuf // Array of pointers to coded output buffers
);
int e = 0;
for (size_t i = 0; i < objcfg.nbchunks; i++)
{
if( pattern[i] )
{
memcpy( stripes[i].buffer, outbuf[e], objcfg.chunksize );
e++;
}
}
}
};
XrdEcRedundancyProvider.hh
Class for computing parities and recovering data.
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdEc::Convert
Definition XrdEcRedundancyProvider.cc:31
XrdEc::Convert::toString
static std::string toString(Args &&...args)
Definition XrdEcRedundancyProvider.cc:40
XrdEc::RedundancyProvider::RedundancyProvider
RedundancyProvider(const ObjCfg &objcfg)
Definition XrdEcRedundancyProvider.cc:151
XrdEc::RedundancyProvider::compute
void compute(stripes_t &stripes)
Definition XrdEcRedundancyProvider.cc:233
XrdCl::stError
const uint16_t stError
An error occurred that could potentially be retried.
Definition XrdClStatus.hh:32
XrdCl::errDataError
const uint16_t errDataError
data is corrupted
Definition XrdClStatus.hh:63
XrdEc
Definition XrdClZipArchive.hh:45
XrdEc::stripes_t
std::vector< stripe_t > stripes_t
All stripes in a block.
Definition XrdEcUtilities.hh:64
XrdEc::gf_gen_decode_matrix
static int gf_gen_decode_matrix(unsigned char *encode_matrix, unsigned char *decode_matrix, unsigned int *decode_index, unsigned char *src_err_list, unsigned char *src_in_err, unsigned int nerrs, unsigned int nsrcerrs, unsigned int k, unsigned int m)
Definition XrdEcRedundancyProvider.cc:73
XrdEc::ObjCfg
Definition XrdEcObjCfg.hh:34
XrdEc::ObjCfg::nbdata
const uint8_t nbdata
Definition XrdEcObjCfg.hh:87
XrdEc::ObjCfg::nbchunks
const uint8_t nbchunks
Definition XrdEcObjCfg.hh:85
XrdEc::ObjCfg::nbparity
const uint8_t nbparity
Definition XrdEcObjCfg.hh:86