[ceph.git] / ceph / src / erasure-code / ErasureCode.cc

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
// vim: ts=8 sw=2 smarttab
/*
 * Ceph distributed storage system
 *
 * Copyright (C) 2014 Cloudwatt <libre.licensing@cloudwatt.com>
 * Copyright (C) 2014 Red Hat <contact@redhat.com>
 *
 * Author: Loic Dachary <loic@dachary.org>
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 * 
 */

#include <algorithm>
#include <cerrno>

#include "ErasureCode.h"

#include "common/strtol.h"
#include "include/buffer.h"
#include "crush/CrushWrapper.h"
#include "osd/osd_types.h"

#define DEFAULT_RULE_ROOT "default"
#define DEFAULT_RULE_FAILURE_DOMAIN "host"

using std::make_pair;
using std::map;
using std::ostream;
using std::pair;
using std::set;
using std::string;
using std::vector;

using ceph::bufferlist;

namespace ceph {
const unsigned ErasureCode::SIMD_ALIGN = 32;

int ErasureCode::init(
  ErasureCodeProfile &profile,
  std::ostream *ss)
{
  int err = 0;
  err |= to_string("crush-root", profile,
		   &rule_root,
		   DEFAULT_RULE_ROOT, ss);
  err |= to_string("crush-failure-domain", profile,
		   &rule_failure_domain,
		   DEFAULT_RULE_FAILURE_DOMAIN, ss);
  err |= to_string("crush-device-class", profile,
		   &rule_device_class,
		   "", ss);
  if (err)
    return err;
  _profile = profile;
  return 0;
}

int ErasureCode::create_rule(
  const std::string &name,
  CrushWrapper &crush,
  std::ostream *ss) const
{
  int ruleid = crush.add_simple_rule(
    name,
    rule_root,
    rule_failure_domain,
    rule_device_class,
    "indep",
    pg_pool_t::TYPE_ERASURE,
    ss);

  if (ruleid < 0)
    return ruleid;

  crush.set_rule_mask_max_size(ruleid, get_chunk_count());
  return ruleid;
}

int ErasureCode::sanity_check_k_m(int k, int m, ostream *ss)
{
  if (k < 2) {
    *ss << "k=" << k << " must be >= 2" << std::endl;
    return -EINVAL;
  }
  if (m < 1) {
    *ss << "m=" << m << " must be >= 1" << std::endl;
    return -EINVAL;
  }
  return 0;
}

int ErasureCode::chunk_index(unsigned int i) const
{
  return chunk_mapping.size() > i ? chunk_mapping[i] : i;
}

int ErasureCode::_minimum_to_decode(const set<int> &want_to_read,
                                   const set<int> &available_chunks,
                                   set<int> *minimum)
{
  if (includes(available_chunks.begin(), available_chunks.end(),
	       want_to_read.begin(), want_to_read.end())) {
    *minimum = want_to_read;
  } else {
    unsigned int k = get_data_chunk_count();
    if (available_chunks.size() < (unsigned)k)
      return -EIO;
    set<int>::iterator i;
    unsigned j;
    for (i = available_chunks.begin(), j = 0; j < (unsigned)k; ++i, j++)
      minimum->insert(*i);
  }
  return 0;
}

int ErasureCode::minimum_to_decode(const set<int> &want_to_read,
                                   const set<int> &available_chunks,
                                   map<int, vector<pair<int, int>>> *minimum)
{
  set<int> minimum_shard_ids;
  int r = _minimum_to_decode(want_to_read, available_chunks, &minimum_shard_ids);
  if (r != 0) {
    return r;
  }
  vector<pair<int, int>> default_subchunks;
  default_subchunks.push_back(make_pair(0, get_sub_chunk_count()));
  for (auto &&id : minimum_shard_ids) {
    minimum->insert(make_pair(id, default_subchunks));
  }
  return 0;
}

int ErasureCode::minimum_to_decode_with_cost(const set<int> &want_to_read,
                                             const map<int, int> &available,
                                             set<int> *minimum)
{
  set <int> available_chunks;
  for (map<int, int>::const_iterator i = available.begin();
       i != available.end();
       ++i)
    available_chunks.insert(i->first);
  return _minimum_to_decode(want_to_read, available_chunks, minimum);
}

int ErasureCode::encode_prepare(const bufferlist &raw,
                                map<int, bufferlist> &encoded) const
{
  unsigned int k = get_data_chunk_count();
  unsigned int m = get_chunk_count() - k;
  unsigned blocksize = get_chunk_size(raw.length());
  unsigned padded_chunks = k - raw.length() / blocksize;
  bufferlist prepared = raw;

  for (unsigned int i = 0; i < k - padded_chunks; i++) {
    bufferlist &chunk = encoded[chunk_index(i)];
    chunk.substr_of(prepared, i * blocksize, blocksize);
    chunk.rebuild_aligned_size_and_memory(blocksize, SIMD_ALIGN);
    ceph_assert(chunk.is_contiguous());
  }
  if (padded_chunks) {
    unsigned remainder = raw.length() - (k - padded_chunks) * blocksize;
    bufferptr buf(buffer::create_aligned(blocksize, SIMD_ALIGN));

    raw.begin((k - padded_chunks) * blocksize).copy(remainder, buf.c_str());
    buf.zero(remainder, blocksize - remainder);
    encoded[chunk_index(k-padded_chunks)].push_back(std::move(buf));

    for (unsigned int i = k - padded_chunks + 1; i < k; i++) {
      bufferptr buf(buffer::create_aligned(blocksize, SIMD_ALIGN));
      buf.zero();
      encoded[chunk_index(i)].push_back(std::move(buf));
    }
  }
  for (unsigned int i = k; i < k + m; i++) {
    bufferlist &chunk = encoded[chunk_index(i)];
    chunk.push_back(buffer::create_aligned(blocksize, SIMD_ALIGN));
  }

  return 0;
}

int ErasureCode::encode(const set<int> &want_to_encode,
                        const bufferlist &in,
                        map<int, bufferlist> *encoded)
{
  unsigned int k = get_data_chunk_count();
  unsigned int m = get_chunk_count() - k;
  bufferlist out;
  int err = encode_prepare(in, *encoded);
  if (err)
    return err;
  encode_chunks(want_to_encode, encoded);
  for (unsigned int i = 0; i < k + m; i++) {
    if (want_to_encode.count(i) == 0)
      encoded->erase(i);
  }
  return 0;
}

int ErasureCode::encode_chunks(const set<int> &want_to_encode,
                               map<int, bufferlist> *encoded)
{
  ceph_abort_msg("ErasureCode::encode_chunks not implemented");
}
 
int ErasureCode::_decode(const set<int> &want_to_read,
			 const map<int, bufferlist> &chunks,
			 map<int, bufferlist> *decoded)
{
  vector<int> have;
  have.reserve(chunks.size());
  for (map<int, bufferlist>::const_iterator i = chunks.begin();
       i != chunks.end();
       ++i) {
    have.push_back(i->first);
  }
  if (includes(
	have.begin(), have.end(), want_to_read.begin(), want_to_read.end())) {
    for (set<int>::iterator i = want_to_read.begin();
	 i != want_to_read.end();
	 ++i) {
      (*decoded)[*i] = chunks.find(*i)->second;
    }
    return 0;
  }
  unsigned int k = get_data_chunk_count();
  unsigned int m = get_chunk_count() - k;
  unsigned blocksize = (*chunks.begin()).second.length();
  for (unsigned int i =  0; i < k + m; i++) {
    if (chunks.find(i) == chunks.end()) {
      bufferlist tmp;
      bufferptr ptr(buffer::create_aligned(blocksize, SIMD_ALIGN));
      tmp.push_back(ptr);
      tmp.claim_append((*decoded)[i]);
      (*decoded)[i].swap(tmp);
    } else {
      (*decoded)[i] = chunks.find(i)->second;
      (*decoded)[i].rebuild_aligned(SIMD_ALIGN);
    }
  }
  return decode_chunks(want_to_read, chunks, decoded);
}

int ErasureCode::decode(const set<int> &want_to_read,
                        const map<int, bufferlist> &chunks,
                        map<int, bufferlist> *decoded, int chunk_size)
{
  return _decode(want_to_read, chunks, decoded);
}

int ErasureCode::decode_chunks(const set<int> &want_to_read,
                               const map<int, bufferlist> &chunks,
                               map<int, bufferlist> *decoded)
{
  ceph_abort_msg("ErasureCode::decode_chunks not implemented");
}

int ErasureCode::parse(const ErasureCodeProfile &profile,
		       ostream *ss)
{
  return to_mapping(profile, ss);
}

const vector<int> &ErasureCode::get_chunk_mapping() const {
  return chunk_mapping;
}

int ErasureCode::to_mapping(const ErasureCodeProfile &profile,
			    ostream *ss)
{
  if (profile.find("mapping") != profile.end()) {
    std::string mapping = profile.find("mapping")->second;
    int position = 0;
    vector<int> coding_chunk_mapping;
    for(std::string::iterator it = mapping.begin(); it != mapping.end(); ++it) {
      if (*it == 'D')
	chunk_mapping.push_back(position);
      else
	coding_chunk_mapping.push_back(position);
      position++;
    }
    chunk_mapping.insert(chunk_mapping.end(),
			 coding_chunk_mapping.begin(),
			 coding_chunk_mapping.end());
  }
  return 0;
}

int ErasureCode::to_int(const std::string &name,
			ErasureCodeProfile &profile,
			int *value,
			const std::string &default_value,
			ostream *ss)
{
  if (profile.find(name) == profile.end() ||
      profile.find(name)->second.size() == 0)
    profile[name] = default_value;
  std::string p = profile.find(name)->second;
  std::string err;
  int r = strict_strtol(p.c_str(), 10, &err);
  if (!err.empty()) {
    *ss << "could not convert " << name << "=" << p
	<< " to int because " << err
	<< ", set to default " << default_value << std::endl;
    *value = strict_strtol(default_value.c_str(), 10, &err);
    return -EINVAL;
  }
  *value = r;
  return 0;
}

int ErasureCode::to_bool(const std::string &name,
			 ErasureCodeProfile &profile,
			 bool *value,
			 const std::string &default_value,
			 ostream *ss)
{
  if (profile.find(name) == profile.end() ||
      profile.find(name)->second.size() == 0)
    profile[name] = default_value;
  const std::string p = profile.find(name)->second;
  *value = (p == "yes") || (p == "true");
  return 0;
}

int ErasureCode::to_string(const std::string &name,
			   ErasureCodeProfile &profile,
			   std::string *value,
			   const std::string &default_value,
			   ostream *ss)
{
  if (profile.find(name) == profile.end() ||
      profile.find(name)->second.size() == 0)
    profile[name] = default_value;
  *value = profile[name];
  return 0;
}

int ErasureCode::decode_concat(const map<int, bufferlist> &chunks,
			       bufferlist *decoded)
{
  set<int> want_to_read;

  for (unsigned int i = 0; i < get_data_chunk_count(); i++) {
    want_to_read.insert(chunk_index(i));
  }
  map<int, bufferlist> decoded_map;
  int r = _decode(want_to_read, chunks, &decoded_map);
  if (r == 0) {
    for (unsigned int i = 0; i < get_data_chunk_count(); i++) {
      decoded->claim_append(decoded_map[chunk_index(i)]);
    }
  }
  return r;
}
}
Commit	Line	Data
7c673cae FG	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3	/*
	4	* Ceph distributed storage system
	5	*
	6	* Copyright (C) 2014 Cloudwatt <libre.licensing@cloudwatt.com>
	7	* Copyright (C) 2014 Red Hat <contact@redhat.com>
	8	*
	9	* Author: Loic Dachary <loic@dachary.org>
	10	*
	11	* This library is free software; you can redistribute it and/or
	12	* modify it under the terms of the GNU Lesser General Public
	13	* License as published by the Free Software Foundation; either
	14	* version 2.1 of the License, or (at your option) any later version.
	15	*
	16	*/
	17
7c673cae	18	#include <algorithm>
9f95a23c	19	#include <cerrno>
7c673cae	20
7c673cae	21	#include "ErasureCode.h"
31f18b77 FG	22
31f18b77 FG	23	#include "common/strtol.h"
7c673cae	24	#include "include/buffer.h"
224ce89b WB	25	#include "crush/CrushWrapper.h"
224ce89b WB	26	#include "osd/osd_types.h"
7c673cae	27
224ce89b WB	28	#define DEFAULT_RULE_ROOT "default"
	29	#define DEFAULT_RULE_FAILURE_DOMAIN "host"
	30
9f95a23c TL	31	using std::make_pair;
	32	using std::map;
	33	using std::ostream;
	34	using std::pair;
	35	using std::set;
	36	using std::string;
	37	using std::vector;
	38
	39	using ceph::bufferlist;
	40
	41	namespace ceph {
11fdf7f2 TL	42	const unsigned ErasureCode::SIMD_ALIGN = 32;
11fdf7f2 TL	43
224ce89b WB	44	int ErasureCode::init(
	45	ErasureCodeProfile &profile,
	46	std::ostream *ss)
	47	{
	48	int err = 0;
	49	err \|= to_string("crush-root", profile,
	50	&rule_root,
	51	DEFAULT_RULE_ROOT, ss);
	52	err \|= to_string("crush-failure-domain", profile,
	53	&rule_failure_domain,
	54	DEFAULT_RULE_FAILURE_DOMAIN, ss);
	55	err \|= to_string("crush-device-class", profile,
	56	&rule_device_class,
	57	"", ss);
	58	if (err)
	59	return err;
	60	_profile = profile;
	61	return 0;
	62	}
	63
	64	int ErasureCode::create_rule(
	65	const std::string &name,
	66	CrushWrapper &crush,
	67	std::ostream *ss) const
	68	{
	69	int ruleid = crush.add_simple_rule(
	70	name,
	71	rule_root,
	72	rule_failure_domain,
	73	rule_device_class,
	74	"indep",
	75	pg_pool_t::TYPE_ERASURE,
	76	ss);
	77
	78	if (ruleid < 0)
	79	return ruleid;
	80
	81	crush.set_rule_mask_max_size(ruleid, get_chunk_count());
	82	return ruleid;
	83	}
	84
11fdf7f2	85	int ErasureCode::sanity_check_k_m(int k, int m, ostream *ss)
7c673cae FG	86	{
	87	if (k < 2) {
	88	*ss << "k=" << k << " must be >= 2" << std::endl;
	89	return -EINVAL;
7c673cae	90	}
11fdf7f2 TL	91	if (m < 1) {
	92	*ss << "m=" << m << " must be >= 1" << std::endl;
	93	return -EINVAL;
	94	}
	95	return 0;
7c673cae FG	96	}
	97
	98	int ErasureCode::chunk_index(unsigned int i) const
	99	{
	100	return chunk_mapping.size() > i ? chunk_mapping[i] : i;
	101	}
	102
11fdf7f2	103	int ErasureCode::_minimum_to_decode(const set<int> &want_to_read,
7c673cae FG	104	const set<int> &available_chunks,
	105	set<int> *minimum)
	106	{
	107	if (includes(available_chunks.begin(), available_chunks.end(),
	108	want_to_read.begin(), want_to_read.end())) {
	109	*minimum = want_to_read;
	110	} else {
	111	unsigned int k = get_data_chunk_count();
	112	if (available_chunks.size() < (unsigned)k)
	113	return -EIO;
	114	set<int>::iterator i;
	115	unsigned j;
	116	for (i = available_chunks.begin(), j = 0; j < (unsigned)k; ++i, j++)
	117	minimum->insert(*i);
	118	}
	119	return 0;
	120	}
	121
11fdf7f2 TL	122	int ErasureCode::minimum_to_decode(const set<int> &want_to_read,
	123	const set<int> &available_chunks,
	124	map<int, vector<pair<int, int>>> *minimum)
	125	{
	126	set<int> minimum_shard_ids;
	127	int r = _minimum_to_decode(want_to_read, available_chunks, &minimum_shard_ids);
	128	if (r != 0) {
	129	return r;
	130	}
	131	vector<pair<int, int>> default_subchunks;
	132	default_subchunks.push_back(make_pair(0, get_sub_chunk_count()));
	133	for (auto &&id : minimum_shard_ids) {
	134	minimum->insert(make_pair(id, default_subchunks));
	135	}
	136	return 0;
	137	}
	138
7c673cae FG	139	int ErasureCode::minimum_to_decode_with_cost(const set<int> &want_to_read,
	140	const map<int, int> &available,
	141	set<int> *minimum)
	142	{
	143	set <int> available_chunks;
	144	for (map<int, int>::const_iterator i = available.begin();
	145	i != available.end();
	146	++i)
	147	available_chunks.insert(i->first);
11fdf7f2	148	return _minimum_to_decode(want_to_read, available_chunks, minimum);
7c673cae FG	149	}
	150
	151	int ErasureCode::encode_prepare(const bufferlist &raw,
	152	map<int, bufferlist> &encoded) const
	153	{
	154	unsigned int k = get_data_chunk_count();
	155	unsigned int m = get_chunk_count() - k;
	156	unsigned blocksize = get_chunk_size(raw.length());
	157	unsigned padded_chunks = k - raw.length() / blocksize;
	158	bufferlist prepared = raw;
	159
	160	for (unsigned int i = 0; i < k - padded_chunks; i++) {
	161	bufferlist &chunk = encoded[chunk_index(i)];
	162	chunk.substr_of(prepared, i * blocksize, blocksize);
	163	chunk.rebuild_aligned_size_and_memory(blocksize, SIMD_ALIGN);
11fdf7f2	164	ceph_assert(chunk.is_contiguous());
7c673cae FG	165	}
	166	if (padded_chunks) {
	167	unsigned remainder = raw.length() - (k - padded_chunks) * blocksize;
	168	bufferptr buf(buffer::create_aligned(blocksize, SIMD_ALIGN));
	169
9f95a23c	170	raw.begin((k - padded_chunks) * blocksize).copy(remainder, buf.c_str());
7c673cae FG	171	buf.zero(remainder, blocksize - remainder);
	172	encoded[chunk_index(k-padded_chunks)].push_back(std::move(buf));
	173
	174	for (unsigned int i = k - padded_chunks + 1; i < k; i++) {
	175	bufferptr buf(buffer::create_aligned(blocksize, SIMD_ALIGN));
	176	buf.zero();
	177	encoded[chunk_index(i)].push_back(std::move(buf));
	178	}
	179	}
	180	for (unsigned int i = k; i < k + m; i++) {
	181	bufferlist &chunk = encoded[chunk_index(i)];
	182	chunk.push_back(buffer::create_aligned(blocksize, SIMD_ALIGN));
	183	}
	184
	185	return 0;
	186	}
	187
	188	int ErasureCode::encode(const set<int> &want_to_encode,
	189	const bufferlist &in,
	190	map<int, bufferlist> *encoded)
	191	{
	192	unsigned int k = get_data_chunk_count();
	193	unsigned int m = get_chunk_count() - k;
	194	bufferlist out;
	195	int err = encode_prepare(in, *encoded);
	196	if (err)
	197	return err;
	198	encode_chunks(want_to_encode, encoded);
	199	for (unsigned int i = 0; i < k + m; i++) {
	200	if (want_to_encode.count(i) == 0)
	201	encoded->erase(i);
	202	}
	203	return 0;
	204	}
	205
	206	int ErasureCode::encode_chunks(const set<int> &want_to_encode,
	207	map<int, bufferlist> *encoded)
	208	{
11fdf7f2	209	ceph_abort_msg("ErasureCode::encode_chunks not implemented");
7c673cae FG	210	}
7c673cae FG	211
11fdf7f2 TL	212	int ErasureCode::_decode(const set<int> &want_to_read,
	213	const map<int, bufferlist> &chunks,
	214	map<int, bufferlist> *decoded)
7c673cae FG	215	{
	216	vector<int> have;
	217	have.reserve(chunks.size());
	218	for (map<int, bufferlist>::const_iterator i = chunks.begin();
	219	i != chunks.end();
	220	++i) {
	221	have.push_back(i->first);
	222	}
	223	if (includes(
	224	have.begin(), have.end(), want_to_read.begin(), want_to_read.end())) {
	225	for (set<int>::iterator i = want_to_read.begin();
	226	i != want_to_read.end();
	227	++i) {
	228	(decoded)[i] = chunks.find(*i)->second;
	229	}
	230	return 0;
	231	}
	232	unsigned int k = get_data_chunk_count();
	233	unsigned int m = get_chunk_count() - k;
	234	unsigned blocksize = (*chunks.begin()).second.length();
	235	for (unsigned int i = 0; i < k + m; i++) {
	236	if (chunks.find(i) == chunks.end()) {
11fdf7f2	237	bufferlist tmp;
7c673cae	238	bufferptr ptr(buffer::create_aligned(blocksize, SIMD_ALIGN));
11fdf7f2 TL	239	tmp.push_back(ptr);
	240	tmp.claim_append((*decoded)[i]);
	241	(*decoded)[i].swap(tmp);
7c673cae FG	242	} else {
	243	(*decoded)[i] = chunks.find(i)->second;
	244	(*decoded)[i].rebuild_aligned(SIMD_ALIGN);
	245	}
	246	}
	247	return decode_chunks(want_to_read, chunks, decoded);
	248	}
	249
11fdf7f2 TL	250	int ErasureCode::decode(const set<int> &want_to_read,
	251	const map<int, bufferlist> &chunks,
	252	map<int, bufferlist> *decoded, int chunk_size)
	253	{
	254	return _decode(want_to_read, chunks, decoded);
	255	}
	256
7c673cae FG	257	int ErasureCode::decode_chunks(const set<int> &want_to_read,
	258	const map<int, bufferlist> &chunks,
	259	map<int, bufferlist> *decoded)
	260	{
11fdf7f2	261	ceph_abort_msg("ErasureCode::decode_chunks not implemented");
7c673cae FG	262	}
	263
	264	int ErasureCode::parse(const ErasureCodeProfile &profile,
	265	ostream *ss)
	266	{
	267	return to_mapping(profile, ss);
	268	}
	269
	270	const vector<int> &ErasureCode::get_chunk_mapping() const {
	271	return chunk_mapping;
	272	}
	273
	274	int ErasureCode::to_mapping(const ErasureCodeProfile &profile,
	275	ostream *ss)
	276	{
	277	if (profile.find("mapping") != profile.end()) {
	278	std::string mapping = profile.find("mapping")->second;
	279	int position = 0;
	280	vector<int> coding_chunk_mapping;
	281	for(std::string::iterator it = mapping.begin(); it != mapping.end(); ++it) {
	282	if (*it == 'D')
	283	chunk_mapping.push_back(position);
	284	else
	285	coding_chunk_mapping.push_back(position);
	286	position++;
	287	}
	288	chunk_mapping.insert(chunk_mapping.end(),
	289	coding_chunk_mapping.begin(),
	290	coding_chunk_mapping.end());
	291	}
	292	return 0;
	293	}
	294
	295	int ErasureCode::to_int(const std::string &name,
	296	ErasureCodeProfile &profile,
	297	int *value,
	298	const std::string &default_value,
	299	ostream *ss)
	300	{
	301	if (profile.find(name) == profile.end() \|\|
	302	profile.find(name)->second.size() == 0)
	303	profile[name] = default_value;
	304	std::string p = profile.find(name)->second;
	305	std::string err;
	306	int r = strict_strtol(p.c_str(), 10, &err);
	307	if (!err.empty()) {
	308	*ss << "could not convert " << name << "=" << p
	309	<< " to int because " << err
	310	<< ", set to default " << default_value << std::endl;
	311	*value = strict_strtol(default_value.c_str(), 10, &err);
	312	return -EINVAL;
	313	}
	314	*value = r;
	315	return 0;
	316	}
	317
	318	int ErasureCode::to_bool(const std::string &name,
	319	ErasureCodeProfile &profile,
	320	bool *value,
	321	const std::string &default_value,
	322	ostream *ss)
	323	{
	324	if (profile.find(name) == profile.end() \|\|
	325	profile.find(name)->second.size() == 0)
326	profile[name] = default_value;
327	const std::string p = profile.find(name)->second;
328	*value = (p == "yes") \|\| (p == "true");
329	return 0;
330	}
331
332	int ErasureCode::to_string(const std::string &name,
333	ErasureCodeProfile &profile,
334	std::string *value,
335	const std::string &default_value,
336	ostream *ss)
337	{
338	if (profile.find(name) == profile.end() \|\|
339	profile.find(name)->second.size() == 0)
340	profile[name] = default_value;
341	*value = profile[name];
342	return 0;
343	}
344
345	int ErasureCode::decode_concat(const map<int, bufferlist> &chunks,
346	bufferlist *decoded)
347	{
348	set<int> want_to_read;
349
350	for (unsigned int i = 0; i < get_data_chunk_count(); i++) {
351	want_to_read.insert(chunk_index(i));
352	}
353	map<int, bufferlist> decoded_map;
11fdf7f2	354	int r = _decode(want_to_read, chunks, &decoded_map);
7c673cae FG	355	if (r == 0) {
	356	for (unsigned int i = 0; i < get_data_chunk_count(); i++) {
	357	decoded->claim_append(decoded_map[chunk_index(i)]);
	358	}
	359	}
	360	return r;
	361	}
9f95a23c	362	}