Subject: [Boost-commit] svn:boost r55190 - sandbox/boost/mapreduce/intermediates
From: cdm.henderson_at_[hidden]
Date: 2009-07-30 14:43:03
Author: chenderson
Date: 2009-07-26 17:21:07 EDT (Sun, 26 Jul 2009)
New Revision: 55190
URL: http://svn.boost.org/trac/boost/changeset/55190
Log:
Added missing file
Added:
sandbox/boost/mapreduce/intermediates/local_disk.hpp (contents, props changed)
Added: sandbox/boost/mapreduce/intermediates/local_disk.hpp
==============================================================================
--- (empty file)
+++ sandbox/boost/mapreduce/intermediates/local_disk.hpp 2009-07-26 17:21:07 EDT (Sun, 26 Jul 2009)
@@ -0,0 +1,290 @@
+// Boost.MapReduce library
+//
+// Copyright (C) 2009 Craig Henderson.
+// cdm.henderson_at_[hidden]
+//
+// Use, modification and distribution is subject to the
+// Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+// For more information, see http://www.boost.org/libs/mapreduce/
+//
+
+#ifndef BOOST_MAPREDUCE_LOCAL_DISK_INTERMEDIATES_HPP
+#define BOOST_MAPREDUCE_LOCAL_DISK_INTERMEDIATES_HPP
+
+#include <iomanip> // setw
+#include <fstream> // linux
+#include <boost/unordered_map.hpp>
+#include <boost/mapreduce/hash_partitioner.hpp>
+
+namespace boost {
+
+namespace mapreduce {
+
+struct null_combiner;
+
+namespace detail {
+
+struct file_merger
+{
+ template<typename List>
+ void operator()(List const &filenames, char const *dest)
+ {
+ buffer_.reset(new char[buffer_size_]);
+
+ std::ofstream outfile(dest, std::ios_base::in | std::ios_base::binary);
+ for (typename List::const_iterator it=filenames.begin(); it!=filenames.end(); ++it)
+ copy_file(*it, outfile);
+ }
+
+ private:
+ void copy_file(std::string const &filename, std::ofstream &outfile)
+ {
+ boost::uint64_t remaining = boost::filesystem::file_size(filename);
+
+ std::ifstream infile(filename.c_str(), std::ios_base::in | std::ios_base::binary);
+ if (!infile.is_open())
+ {
+ std::ostringstream err;
+ err << "Failed to open file: " << filename;
+ BOOST_THROW_EXCEPTION(std::runtime_error(err.str()));
+ }
+
+ std::streamsize read_write_size = buffer_size_;
+ while (remaining > 0)
+ {
+ if (remaining < (boost::uint64_t)read_write_size)
+ read_write_size = (std::streamsize)remaining;
+
+ infile.read(buffer_.get(), read_write_size);
+ if (infile.bad() || infile.fail())
+ {
+ std::ostringstream err;
+ err << "Error reading file " << filename;
+ BOOST_THROW_EXCEPTION(std::runtime_error(err.str()));
+ }
+
+ outfile.write(buffer_.get(), read_write_size);
+ if (outfile.bad() || outfile.fail())
+ {
+ std::ostringstream err;
+ err << "Error writing to file";
+ BOOST_THROW_EXCEPTION(std::runtime_error(err.str()));
+ }
+
+ remaining -= read_write_size;
+ }
+ }
+
+ std::auto_ptr<char> buffer_;
+ static unsigned const buffer_size_ = 1048576;
+};
+
+struct file_sorter
+{
+ bool const operator()(char const *in, char const *out, unsigned const offset) const
+ {
+ return mapreduce::merge_sort(in, out, offset);
+ }
+};
+
+} // namespace detail
+
+namespace intermediates {
+
+template<
+ typename MapTask,
+ typename SortFn=mapreduce::detail::file_sorter,
+ typename MergeFn=mapreduce::detail::file_merger,
+ typename PartitionFn=mapreduce::hash_partitioner>
+class local_disk : boost::noncopyable
+{
+ private:
+ typedef
+ boost::unordered_map<
+ size_t, // hash value of intermediate key (R)
+ std::pair<
+ std::string, // filename
+ boost::shared_ptr<std::ofstream> > > // file stream
+ intermediates_t;
+
+ public:
+ typedef MapTask map_task_type;
+ typedef typename intermediates_t::const_iterator const_iterator;
+
+ local_disk(unsigned const num_partitions)
+ : num_partitions_(num_partitions)
+ {
+ }
+
+ ~local_disk()
+ {
+ try
+ {
+ this->close_files();
+
+ // delete the temporary files
+ for (intermediates_t::iterator it=intermediates_.begin(); it!=intermediates_.end(); ++it)
+ boost::filesystem::remove(it->second.first);
+ }
+ catch (std::exception const &/*e*/)
+ {
+// std::cerr << "\nError: " << e.what() << "\n";
+ }
+ }
+
+ bool const insert(typename map_task_type::intermediate_key_type const &key,
+ typename map_task_type::intermediate_value_type const &value)
+ {
+ unsigned const partition = partitioner_(key, num_partitions_);
+ intermediates_t::iterator it = intermediates_.insert(make_pair(partition, intermediates_t::mapped_type())).first;
+ if (it->second.first.empty())
+ {
+ it->second.first = platform::get_temporary_filename();
+ it->second.second.reset(new std::ofstream);
+ }
+
+ if (!it->second.second->is_open())
+ it->second.second->open(it->second.first.c_str(), std::ios_base::out | std::ios_base::app);
+ assert(it->second.second->is_open());
+
+ std::ostringstream key_text_stream;
+ key_text_stream << key;
+ std::string key_text(key_text_stream.str());
+ *it->second.second << std::setw(10) << key_text.length() << "\t" << key << "\t" << value << "\n";
+ return !(it->second.second->bad() || it->second.second->fail());
+ }
+
+ bool const get_partition_filename(size_t const partition, std::string &filename) const
+ {
+ intermediates_t::const_iterator it=completed_intermediates_.find(partition);
+ if (it == completed_intermediates_.end())
+ return false;
+ filename = it->second.first;
+ return true;
+ }
+
+ template<typename FnObj>
+ void combine(FnObj &fn_obj)
+ {
+ this->close_files();
+ for (intermediates_t::iterator it=intermediates_.begin(); it!=intermediates_.end(); ++it)
+ {
+ std::string infilename = it->second.first;
+ std::string outfilename = platform::get_temporary_filename();
+
+ // sort the input file
+ SortFn()(infilename.c_str(), outfilename.c_str(), 11);
+ boost::filesystem::remove(infilename);
+ std::swap(infilename, outfilename);
+
+ std::string key, last_key;
+ std::ifstream infile(infilename.c_str());
+ while (!infile.eof())
+ {
+ typename map_task_type::intermediate_value_type value;
+ if (read_record(infile, key, value))
+ {
+ if (key != last_key && key.length() > 0)
+ {
+ if (last_key.length() > 0)
+ fn_obj.finish(last_key, *this);
+ if (key.length() > 0)
+ {
+ fn_obj.start(key);
+ std::swap(key, last_key);
+ }
+ }
+
+ fn_obj(value);
+ }
+ }
+
+ if (last_key.length() > 0)
+ fn_obj.finish(last_key, *this);
+
+ infile.close();
+
+ boost::filesystem::remove(infilename);
+ }
+ this->close_files();
+
+ assert(completed_intermediates_.size() == 0);
+ std::swap(completed_intermediates_, intermediates_);
+ }
+
+ void combine(mapreduce::null_combiner &/*fn_obj*/)
+ {
+ this->close_files();
+ assert(completed_intermediates_.size() == 0);
+ std::swap(completed_intermediates_, intermediates_);
+ }
+
+ static bool const read_record(std::ifstream &infile,
+ typename map_task_type::intermediate_key_type &key,
+ typename map_task_type::intermediate_value_type &value)
+ {
+#if defined(__SGI_STL_PORT)
+ size_t keylen;
+#else
+ std::streamsize keylen;
+#endif
+ keylen = 0;
+ infile >> keylen;
+ if (infile.eof() || keylen == 0)
+ return false;
+
+ char tab;
+ infile.read(&tab, 1);
+
+ key.resize(keylen);
+ infile.read(&*key.begin(), keylen);
+ infile.read(&tab, 1);
+ infile >> value;
+ return true;
+ }
+
+ template<typename List>
+ static std::string merge_and_sort(List const &filenames)
+ {
+ assert(filenames.size() > 0);
+
+ // first merge
+ std::string const temp = platform::get_temporary_filename();
+ MergeFn()(filenames, temp.c_str());
+
+ // then sort
+ typename List::const_iterator it=filenames.begin();
+ SortFn()(temp.c_str(), it->c_str(), 11);
+
+ // delete merge source files
+ boost::filesystem::remove(temp);
+ std::for_each(++it, filenames.end(), boost::bind(detail::delete_file, _1));
+
+ // return the resulting filename
+ return *filenames.begin();
+ }
+
+ private:
+ void close_files(void)
+ {
+ for (intermediates_t::iterator it=intermediates_.begin(); it!=intermediates_.end(); ++it)
+ if (it->second.second && it->second.second->is_open())
+ it->second.second->close();
+ }
+
+ private:
+ unsigned const num_partitions_;
+ intermediates_t intermediates_;
+ intermediates_t completed_intermediates_;
+ PartitionFn partitioner_;
+};
+
+} // namespace intermediates
+
+} // namespace mapreduce
+
+} // namespace boost
+
+#endif // BOOST_MAPREDUCE_LOCAL_DISK_INTERMEDIATES_HPP