Initial import

Author: alex

AUTHORS

@@ -0,0 +1 @@
+Alexander Hurd

COPYING

@@ -0,0 +1 @@
+<Place your desired license here.>

ChangeLog

@@ -0,0 +1,365 @@
+Installation Instructions
+*************************
+
+Copyright (C) 1994, 1995, 1996, 1999, 2000, 2001, 2002, 2004, 2005,
+2006, 2007, 2008, 2009 Free Software Foundation, Inc.
+
+   Copying and distribution of this file, with or without modification,
+are permitted in any medium without royalty provided the copyright
+notice and this notice are preserved.  This file is offered as-is,
+without warranty of any kind.
+
+Basic Installation
+==================
+
+   Briefly, the shell commands `./configure; make; make install' should
+configure, build, and install this package.  The following
+more-detailed instructions are generic; see the `README' file for
+instructions specific to this package.  Some packages provide this
+`INSTALL' file but do not implement all of the features documented
+below.  The lack of an optional feature in a given package is not
+necessarily a bug.  More recommendations for GNU packages can be found
+in *note Makefile Conventions: (standards)Makefile Conventions.
+
+   The `configure' shell script attempts to guess correct values for
+various system-dependent variables used during compilation.  It uses
+those values to create a `Makefile' in each directory of the package.
+It may also create one or more `.h' files containing system-dependent
+definitions.  Finally, it creates a shell script `config.status' that
+you can run in the future to recreate the current configuration, and a
+file `config.log' containing compiler output (useful mainly for
+debugging `configure').
+
+   It can also use an optional file (typically called `config.cache'
+and enabled with `--cache-file=config.cache' or simply `-C') that saves
+the results of its tests to speed up reconfiguring.  Caching is
+disabled by default to prevent problems with accidental use of stale
+cache files.
+
+   If you need to do unusual things to compile the package, please try
+to figure out how `configure' could check whether to do them, and mail
+diffs or instructions to the address given in the `README' so they can
+be considered for the next release.  If you are using the cache, and at
+some point `config.cache' contains results you don't want to keep, you
+may remove or edit it.
+
+   The file `configure.ac' (or `configure.in') is used to create
+`configure' by a program called `autoconf'.  You need `configure.ac' if
+you want to change it or regenerate `configure' using a newer version
+of `autoconf'.
+
+   The simplest way to compile this package is:
+
+  1. `cd' to the directory containing the package's source code and type
+     `./configure' to configure the package for your system.
+
+     Running `configure' might take a while.  While running, it prints
+     some messages telling which features it is checking for.
+
+  2. Type `make' to compile the package.
+
+  3. Optionally, type `make check' to run any self-tests that come with
+     the package, generally using the just-built uninstalled binaries.
+
+  4. Type `make install' to install the programs and any data files and
+     documentation.  When installing into a prefix owned by root, it is
+     recommended that the package be configured and built as a regular
+     user, and only the `make install' phase executed with root
+     privileges.
+
+  5. Optionally, type `make installcheck' to repeat any self-tests, but
+     this time using the binaries in their final installed location.
+     This target does not install anything.  Running this target as a
+     regular user, particularly if the prior `make install' required
+     root privileges, verifies that the installation completed
+     correctly.
+
+  6. You can remove the program binaries and object files from the
+     source code directory by typing `make clean'.  To also remove the
+     files that `configure' created (so you can compile the package for
+     a different kind of computer), type `make distclean'.  There is
+     also a `make maintainer-clean' target, but that is intended mainly
+     for the package's developers.  If you use it, you may have to get
+     all sorts of other programs in order to regenerate files that came
+     with the distribution.
+
+  7. Often, you can also type `make uninstall' to remove the installed
+     files again.  In practice, not all packages have tested that
+     uninstallation works correctly, even though it is required by the
+     GNU Coding Standards.
+
+  8. Some packages, particularly those that use Automake, provide `make
+     distcheck', which can by used by developers to test that all other
+     targets like `make install' and `make uninstall' work correctly.
+     This target is generally not run by end users.
+
+Compilers and Options
+=====================
+
+   Some systems require unusual options for compilation or linking that
+the `configure' script does not know about.  Run `./configure --help'
+for details on some of the pertinent environment variables.
+
+   You can give `configure' initial values for configuration parameters
+by setting variables in the command line or in the environment.  Here
+is an example:
+
+     ./configure CC=c99 CFLAGS=-g LIBS=-lposix
+
+   *Note Defining Variables::, for more details.
+
+Compiling For Multiple Architectures
+====================================
+
+   You can compile the package for more than one kind of computer at the
+same time, by placing the object files for each architecture in their
+own directory.  To do this, you can use GNU `make'.  `cd' to the
+directory where you want the object files and executables to go and run
+the `configure' script.  `configure' automatically checks for the
+source code in the directory that `configure' is in and in `..'.  This
+is known as a "VPATH" build.
+
+   With a non-GNU `make', it is safer to compile the package for one
+architecture at a time in the source code directory.  After you have
+installed the package for one architecture, use `make distclean' before
+reconfiguring for another architecture.
+
+   On MacOS X 10.5 and later systems, you can create libraries and
+executables that work on multiple system types--known as "fat" or
+"universal" binaries--by specifying multiple `-arch' options to the
+compiler but only a single `-arch' option to the preprocessor.  Like
+this:
+
+     ./configure CC="gcc -arch i386 -arch x86_64 -arch ppc -arch ppc64" \
+                 CXX="g++ -arch i386 -arch x86_64 -arch ppc -arch ppc64" \
+                 CPP="gcc -E" CXXCPP="g++ -E"
+
+   This is not guaranteed to produce working output in all cases, you
+may have to build one architecture at a time and combine the results
+using the `lipo' tool if you have problems.
+
+Installation Names
+==================
+
+   By default, `make install' installs the package's commands under
+`/usr/local/bin', include files under `/usr/local/include', etc.  You
+can specify an installation prefix other than `/usr/local' by giving
+`configure' the option `--prefix=PREFIX', where PREFIX must be an
+absolute file name.
+
+   You can specify separate installation prefixes for
+architecture-specific files and architecture-independent files.  If you
+pass the option `--exec-prefix=PREFIX' to `configure', the package uses
+PREFIX as the prefix for installing programs and libraries.
+Documentation and other data files still use the regular prefix.
+
+   In addition, if you use an unusual directory layout you can give
+options like `--bindir=DIR' to specify different values for particular
+kinds of files.  Run `configure --help' for a list of the directories
+you can set and what kinds of files go in them.  In general, the
+default for these options is expressed in terms of `${prefix}', so that
+specifying just `--prefix' will affect all of the other directory
+specifications that were not explicitly provided.
+
+   The most portable way to affect installation locations is to pass the
+correct locations to `configure'; however, many packages provide one or
+both of the following shortcuts of passing variable assignments to the
+`make install' command line to change installation locations without
+having to reconfigure or recompile.
+
+   The first method involves providing an override variable for each
+affected directory.  For example, `make install
+prefix=/alternate/directory' will choose an alternate location for all
+directory configuration variables that were expressed in terms of
+`${prefix}'.  Any directories that were specified during `configure',
+but not in terms of `${prefix}', must each be overridden at install
+time for the entire installation to be relocated.  The approach of
+makefile variable overrides for each directory variable is required by
+the GNU Coding Standards, and ideally causes no recompilation.
+However, some platforms have known limitations with the semantics of
+shared libraries that end up requiring recompilation when using this
+method, particularly noticeable in packages that use GNU Libtool.
+
+   The second method involves providing the `DESTDIR' variable.  For
+example, `make install DESTDIR=/alternate/directory' will prepend
+`/alternate/directory' before all installation names.  The approach of
+`DESTDIR' overrides is not required by the GNU Coding Standards, and
+does not work on platforms that have drive letters.  On the other hand,
+it does better at avoiding recompilation issues, and works well even
+when some directory options were not specified in terms of `${prefix}'
+at `configure' time.
+
+Optional Features
+=================
+
+   If the package supports it, you can cause programs to be installed
+with an extra prefix or suffix on their names by giving `configure' the
+option `--program-prefix=PREFIX' or `--program-suffix=SUFFIX'.
+
+   Some packages pay attention to `--enable-FEATURE' options to
+`configure', where FEATURE indicates an optional part of the package.
+They may also pay attention to `--with-PACKAGE' options, where PACKAGE
+is something like `gnu-as' or `x' (for the X Window System).  The
+`README' should mention any `--enable-' and `--with-' options that the
+package recognizes.
+
+   For packages that use the X Window System, `configure' can usually
+find the X include and library files automatically, but if it doesn't,
+you can use the `configure' options `--x-includes=DIR' and
+`--x-libraries=DIR' to specify their locations.
+
+   Some packages offer the ability to configure how verbose the
+execution of `make' will be.  For these packages, running `./configure
+--enable-silent-rules' sets the default to minimal output, which can be
+overridden with `make V=1'; while running `./configure
+--disable-silent-rules' sets the default to verbose, which can be
+overridden with `make V=0'.
+
+Particular systems
+==================
+
+   On HP-UX, the default C compiler is not ANSI C compatible.  If GNU
+CC is not installed, it is recommended to use the following options in
+order to use an ANSI C compiler:
+
+     ./configure CC="cc -Ae -D_XOPEN_SOURCE=500"
+
+and if that doesn't work, install pre-built binaries of GCC for HP-UX.
+
+   On OSF/1 a.k.a. Tru64, some versions of the default C compiler cannot
+parse its `<wchar.h>' header file.  The option `-nodtk' can be used as
+a workaround.  If GNU CC is not installed, it is therefore recommended
+to try
+
+     ./configure CC="cc"
+
+and if that doesn't work, try
+
+     ./configure CC="cc -nodtk"
+
+   On Solaris, don't put `/usr/ucb' early in your `PATH'.  This
+directory contains several dysfunctional programs; working variants of
+these programs are available in `/usr/bin'.  So, if you need `/usr/ucb'
+in your `PATH', put it _after_ `/usr/bin'.
+
+   On Haiku, software installed for all users goes in `/boot/common',
+not `/usr/local'.  It is recommended to use the following options:
+
+     ./configure --prefix=/boot/common
+
+Specifying the System Type
+==========================
+
+   There may be some features `configure' cannot figure out
+automatically, but needs to determine by the type of machine the package
+will run on.  Usually, assuming the package is built to be run on the
+_same_ architectures, `configure' can figure that out, but if it prints
+a message saying it cannot guess the machine type, give it the
+`--build=TYPE' option.  TYPE can either be a short name for the system
+type, such as `sun4', or a canonical name which has the form:
+
+     CPU-COMPANY-SYSTEM
+
+where SYSTEM can have one of these forms:
+
+     OS
+     KERNEL-OS
+
+   See the file `config.sub' for the possible values of each field.  If
+`config.sub' isn't included in this package, then this package doesn't
+need to know the machine type.
+
+   If you are _building_ compiler tools for cross-compiling, you should
+use the option `--target=TYPE' to select the type of system they will
+produce code for.
+
+   If you want to _use_ a cross compiler, that generates code for a
+platform different from the build platform, you should specify the
+"host" platform (i.e., that on which the generated programs will
+eventually be run) with `--host=TYPE'.
+
+Sharing Defaults
+================
+
+   If you want to set default values for `configure' scripts to share,
+you can create a site shell script called `config.site' that gives
+default values for variables like `CC', `cache_file', and `prefix'.
+`configure' looks for `PREFIX/share/config.site' if it exists, then
+`PREFIX/etc/config.site' if it exists.  Or, you can set the
+`CONFIG_SITE' environment variable to the location of the site script.
+A warning: not all `configure' scripts look for a site script.
+
+Defining Variables
+==================
+
+   Variables not defined in a site shell script can be set in the
+environment passed to `configure'.  However, some packages may run
+configure again during the build, and the customized values of these
+variables may be lost.  In order to avoid this problem, you should set
+them in the `configure' command line, using `VAR=value'.  For example:
+
+     ./configure CC=/usr/local2/bin/gcc
+
+causes the specified `gcc' to be used as the C compiler (unless it is
+overridden in the site shell script).
+
+Unfortunately, this technique does not work for `CONFIG_SHELL' due to
+an Autoconf bug.  Until the bug is fixed you can use this workaround:
+
+     CONFIG_SHELL=/bin/bash /bin/bash ./configure CONFIG_SHELL=/bin/bash
+
+`configure' Invocation
+======================
+
+   `configure' recognizes the following options to control how it
+operates.
+
+`--help'
+`-h'
+     Print a summary of all of the options to `configure', and exit.
+
+`--help=short'
+`--help=recursive'
+     Print a summary of the options unique to this package's
+     `configure', and exit.  The `short' variant lists options used
+     only in the top level, while the `recursive' variant lists options
+     also present in any nested packages.
+
+`--version'
+`-V'
+     Print the version of Autoconf used to generate the `configure'
+     script, and exit.
+
+`--cache-file=FILE'
+     Enable the cache: use and save the results of the tests in FILE,
+     traditionally `config.cache'.  FILE defaults to `/dev/null' to
+     disable caching.
+
+`--config-cache'
+`-C'
+     Alias for `--cache-file=config.cache'.
+
+`--quiet'
+`--silent'
+`-q'
+     Do not print messages saying which checks are being made.  To
+     suppress all normal output, redirect it to `/dev/null' (any error
+     messages will still be shown).
+
+`--srcdir=DIR'
+     Look for the package's source code in directory DIR.  Usually
+     `configure' can determine that directory automatically.
+
+`--prefix=DIR'
+     Use DIR as the installation prefix.  *note Installation Names::
+     for more details, including other options available for fine-tuning
+     the installation locations.
+
+`--no-create'
+`-n'
+     Run the configure checks, but stop before creating any output
+     files.
+
+`configure' also accepts some other, not widely useful, options.  Run
+`configure --help' for more details.
+

INSTALL

@@ -0,0 +1,2 @@
+
+SUBDIRS=src

Makefile.am

@@ -0,0 +1 @@
+Sample NEWS file for qfs_mapred project.

NEWS

@@ -0,0 +1 @@
+Sample readme file for qfs_mapred project.

README

@@ -0,0 +1,14 @@
+dnl Process this file with autoconf to produce a configure script.
+
+AC_PREREQ(2.59)
+AC_INIT(test, 1.0)
+
+
+AC_CANONICAL_SYSTEM
+AM_INIT_AUTOMAKE()
+
+AC_PROG_CXX
+
+AC_CONFIG_FILES(Makefile src/Makefile)
+AC_OUTPUT
+

configure.ac

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+"""A more advanced Mapper, using Python iterators and generators."""
+
+import sys
+import re
+
+def read_input(file):
+    for line in file:
+        # split the line into words
+        yield line.split()
+
+def main(separator='\t'):
+    # input comes from STDIN (standard input)
+    data = read_input(sys.stdin)
+    for words in data:
+        # write the results to STDOUT (standard output);
+        # what we output here will be the input for the
+        # Reduce step, i.e. the input for reducer.py
+        #
+        # tab-delimited; the trivial word count is 1
+        for word in words:
+
+	    if re.match('^[A-Za-z]+$', word):
+		word = word.strip();
+		word = word.upper();
+          	print '%s%s%d' % (word, separator, 1)
+
+if __name__ == "__main__":
+    main()

data/pg4300.txt

@@ -0,0 +1,28 @@
+#!/usr/bin/env python
+"""A more advanced Reducer, using Python iterators and generators."""
+
+from itertools import groupby
+from operator import itemgetter
+import sys
+
+def read_mapper_output(file, separator='\t'):
+    for line in file:
+        yield line.rstrip().split(separator, 1)
+
+def main(separator='\t'):
+    # input comes from STDIN (standard input)
+    data = read_mapper_output(sys.stdin, separator=separator)
+    # groupby groups multiple word-count pairs by word,
+    # and creates an iterator that returns consecutive keys and their group:
+    #   current_word - string containing a word (the key)
+    #   group - iterator yielding all ["<current_word>", "<count>"] items
+    for current_word, group in groupby(data, itemgetter(0)):
+        try:
+            total_count = sum(int(count) for current_word, count in group)
+            print "%s%s%d" % (current_word, separator, total_count)
+        except ValueError:
+            # count was not a number, so silently discard this item
+            pass
+
+if __name__ == "__main__":
+    main()

data/pg5000.txt

@@ -0,0 +1,11 @@
+AM_CPPFLAGS=-Wall -I/home/alex/qfs/build/release/include
+AM_LDFLAGS= -L/home/alex/qfs/build/release/lib -lboost_program_options -lgearman -lqfs_client -lboost_regex -lboost_thread -lpthread
+ 
+bin_PROGRAMS=qfs_mapred_submit mapper_worker mapper_to_qfs_partitions sorter_worker reducer_worker kvsorter
+
+qfs_mapred_submit_SOURCES=qfs_mapred_submit_main.cc
+mapper_worker_SOURCES=mapper_worker_main.cc
+mapper_to_qfs_partitions_SOURCES=mapper_to_qfs_partitions_main.cc
+sorter_worker_SOURCES=sorter_worker_main.cc
+kvsorter_SOURCES=kvsorter_main.cc
+reducer_worker_SOURCES=reducer_worker_main.cc

scripts/mapper.py

@@ -0,0 +1,15 @@
+/*
+ * kv_struct.h
+ *
+ *  Created on: Nov 22, 2012
+ *      Author: alex
+ */
+
+#ifndef KV_STRUCT_H_
+#define KV_STRUCT_H_
+
+struct kv{
+	std::string key;
+	std::string value;
+};
+#endif /* KV_STRUCT_H_ */

scripts/reducer.py

@@ -0,0 +1,17 @@
+/*
+ * mapper_data_struct.h
+ *
+ *  Created on: Nov 23, 2012
+ *      Author: alex
+ */
+
+#ifndef MAPPER_DATA_STRUCT_H_
+#define MAPPER_DATA_STRUCT_H_
+
+struct mapper_data{
+	int partition;
+	std::string data;
+};
+
+
+#endif /* MAPPER_DATA_STRUCT_H_ */

src/Makefile.am

@@ -0,0 +1,41 @@
+/*
+ * workloads.h
+ *
+ *  Created on: Nov 17, 2012
+ *      Author: alex
+ */
+
+#ifndef WORKLOADS_H_
+#define WORKLOADS_H_
+
+#include "json_spirit_headers/json_spirit_value.h"
+
+struct Mapper_Workload
+{
+	std::string job_id;
+	std::string qfs_meta_server_name;
+	unsigned short qfs_meta_server_port;
+	std::string qfs_file_input;
+	std::string qfs_map_folder;
+    std::string python_mapper_function;
+    int partition_count;
+};
+
+struct Sorter_Workload
+{
+	std::string qfs_meta_server_name;
+	unsigned short qfs_meta_server_port;
+	std::string qfs_partition_file_input;
+	std::string qfs_partition_file_output;
+};
+
+struct Reducer_Workload
+{
+	std::string qfs_meta_server_name;
+	unsigned short qfs_meta_server_port;
+	json_spirit::Array qfs_sorted_partition_files;
+	std::string qfs_output_file;
+    std::string python_reducer_function;
+};
+
+#endif /* WORKLOADS_H_ */

src/headers/kv_struct.h

@@ -0,0 +1,54 @@
+#ifndef JSON_SPIRIT_ERROR_POSITION
+#define JSON_SPIRIT_ERROR_POSITION
+
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1020)
+# pragma once
+#endif
+
+#include <string>
+
+namespace json_spirit
+{
+    // An Error_position exception is thrown by the "read_or_throw" functions below on finding an error.
+    // Note the "read_or_throw" functions are around 3 times slower than the standard functions "read" 
+    // functions that return a bool.
+    //
+    struct Error_position
+    {
+        Error_position();
+        Error_position( unsigned int line, unsigned int column, const std::string& reason );
+        bool operator==( const Error_position& lhs ) const;
+        unsigned int line_;
+        unsigned int column_;
+        std::string reason_;
+    };
+
+    inline Error_position::Error_position()
+    :   line_( 0 )
+    ,   column_( 0 )
+    {
+    }
+
+    inline Error_position::Error_position( unsigned int line, unsigned int column, const std::string& reason )
+    :   line_( line )
+    ,   column_( column )
+    ,   reason_( reason )
+    {
+    }
+
+    inline bool Error_position::operator==( const Error_position& lhs ) const
+    {
+        if( this == &lhs ) return true;
+
+        return ( reason_ == lhs.reason_ ) &&
+               ( line_   == lhs.line_ ) &&
+               ( column_ == lhs.column_ ); 
+    }
+}
+
+#endif

src/headers/mapper_data_struct.h

@@ -0,0 +1,137 @@
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#include "json_spirit_reader.h"
+#include "json_spirit_reader_template.h"
+
+using namespace json_spirit;
+
+#ifdef JSON_SPIRIT_VALUE_ENABLED
+    bool json_spirit::read( const std::string& s, Value& value )
+    {
+        return read_string( s, value );
+    }
+    
+    void json_spirit::read_or_throw( const std::string& s, Value& value )
+    {
+        read_string_or_throw( s, value );
+    }
+
+    bool json_spirit::read( std::istream& is, Value& value )
+    {
+        return read_stream( is, value );
+    }
+
+    void json_spirit::read_or_throw( std::istream& is, Value& value )
+    {
+        read_stream_or_throw( is, value );
+    }
+
+    bool json_spirit::read( std::string::const_iterator& begin, std::string::const_iterator end, Value& value )
+    {
+        return read_range( begin, end, value );
+    }
+
+    void json_spirit::read_or_throw( std::string::const_iterator& begin, std::string::const_iterator end, Value& value )
+    {
+        begin = read_range_or_throw( begin, end, value );
+    }
+#endif
+
+#if defined( JSON_SPIRIT_WVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    bool json_spirit::read( const std::wstring& s, wValue& value )
+    {
+        return read_string( s, value );
+    }
+
+    void json_spirit::read_or_throw( const std::wstring& s, wValue& value )
+    {
+        read_string_or_throw( s, value );
+    }
+
+    bool json_spirit::read( std::wistream& is, wValue& value )
+    {
+        return read_stream( is, value );
+    }
+
+    void json_spirit::read_or_throw( std::wistream& is, wValue& value )
+    {
+        read_stream_or_throw( is, value );
+    }
+
+    bool json_spirit::read( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wValue& value )
+    {
+        return read_range( begin, end, value );
+    }
+
+    void json_spirit::read_or_throw( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wValue& value )
+    {
+        begin = read_range_or_throw( begin, end, value );
+    }
+#endif
+
+#ifdef JSON_SPIRIT_MVALUE_ENABLED
+    bool json_spirit::read( const std::string& s, mValue& value )
+    {
+        return read_string( s, value );
+    }
+
+    void json_spirit::read_or_throw( const std::string& s, mValue& value )
+    {
+        read_string_or_throw( s, value );
+    }
+    
+    bool json_spirit::read( std::istream& is, mValue& value )
+    {
+        return read_stream( is, value );
+    }
+
+    void json_spirit::read_or_throw( std::istream& is, mValue& value )
+    {
+        read_stream_or_throw( is, value );
+    }
+
+    bool json_spirit::read( std::string::const_iterator& begin, std::string::const_iterator end, mValue& value )
+    {
+        return read_range( begin, end, value );
+    }
+
+    void json_spirit::read_or_throw( std::string::const_iterator& begin, std::string::const_iterator end, mValue& value )
+    {
+        begin = read_range_or_throw( begin, end, value );
+    }
+#endif
+
+#if defined( JSON_SPIRIT_WMVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    bool json_spirit::read( const std::wstring& s, wmValue& value )
+    {
+        return read_string( s, value );
+    }
+
+    void json_spirit::read_or_throw( const std::wstring& s, wmValue& value )
+    {
+        read_string_or_throw( s, value );
+    }
+
+    bool json_spirit::read( std::wistream& is, wmValue& value )
+    {
+        return read_stream( is, value );
+    }
+
+    void json_spirit::read_or_throw( std::wistream& is, wmValue& value )
+    {
+        read_stream_or_throw( is, value );
+    }
+
+    bool json_spirit::read( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wmValue& value )
+    {
+        return read_range( begin, end, value );
+    }
+
+    void json_spirit::read_or_throw( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wmValue& value )
+    {
+        begin = read_range_or_throw( begin, end, value );
+    }
+#endif

src/headers/workloads.h

@@ -0,0 +1,62 @@
+#ifndef JSON_SPIRIT_READER
+#define JSON_SPIRIT_READER
+
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1020)
+# pragma once
+#endif
+
+#include "json_spirit_value.h"
+#include "json_spirit_error_position.h"
+#include <iostream>
+
+namespace json_spirit
+{
+    // functions to reads a JSON values
+
+#ifdef JSON_SPIRIT_VALUE_ENABLED
+    bool read( const std::string& s, Value& value );
+    bool read( std::istream& is,     Value& value );
+    bool read( std::string::const_iterator& begin, std::string::const_iterator end, Value& value );
+
+    void read_or_throw( const std::string& s, Value& value );  
+    void read_or_throw( std::istream& is,     Value& value );
+    void read_or_throw( std::string::const_iterator& begin, std::string::const_iterator end, Value& value );
+#endif
+
+#if defined( JSON_SPIRIT_WVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    bool read( const std::wstring& s, wValue& value );
+    bool read( std::wistream&  is,    wValue& value );
+    bool read( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wValue& value );    
+
+    void read_or_throw( const std::wstring& s, wValue& value );
+    void read_or_throw( std::wistream& is,     wValue& value );
+    void read_or_throw( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wValue& value );
+#endif
+
+#ifdef JSON_SPIRIT_MVALUE_ENABLED
+    bool read( const std::string& s, mValue& value );
+    bool read( std::istream& is,     mValue& value );
+    bool read( std::string::const_iterator& begin, std::string::const_iterator end, mValue& value );
+
+    void read_or_throw( const std::string& s, mValue& value );  
+    void read_or_throw( std::istream& is,     mValue& value );
+    void read_or_throw( std::string::const_iterator& begin, std::string::const_iterator end, mValue& value );
+#endif
+
+#if defined( JSON_SPIRIT_WMVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    bool read( const std::wstring& s, wmValue& value );
+    bool read( std::wistream& is,     wmValue& value );
+    bool read( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wmValue& value );    
+
+    void read_or_throw( const std::wstring& s, wmValue& value );
+    void read_or_throw( std::wistream& is,     wmValue& value );
+    void read_or_throw( std::wstring::const_iterator& begin, std::wstring::const_iterator end, wmValue& value );
+#endif
+}
+
+#endif

src/json_spirit_headers/json_spirit_error_position.h

@@ -0,0 +1,70 @@
+#ifndef JSON_SPIRIT_READ_STREAM
+#define JSON_SPIRIT_READ_STREAM
+
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1020)
+# pragma once
+#endif
+
+#include "json_spirit_reader_template.h"
+
+namespace json_spirit
+{
+    // these classes allows you to read multiple top level contiguous values from a stream,
+    // the normal stream read functions have a bug that prevent multiple top level values 
+    // from being read unless they are separated by spaces
+
+    template< class Istream_type, class Value_type >
+    class Stream_reader
+    {
+    public:
+
+        Stream_reader( Istream_type& is )
+        :   iters_( is )
+        {
+        }
+
+        bool read_next( Value_type& value )
+        {
+            return read_range( iters_.begin_, iters_.end_, value );
+        }
+
+    private:
+
+        typedef Multi_pass_iters< Istream_type > Mp_iters;
+
+        Mp_iters iters_;
+    };
+
+    template< class Istream_type, class Value_type >
+    class Stream_reader_thrower
+    {
+    public:
+
+        Stream_reader_thrower( Istream_type& is )
+        :   iters_( is )
+        ,    posn_begin_( iters_.begin_, iters_.end_ )
+        ,    posn_end_( iters_.end_, iters_.end_ )
+        {
+        }
+
+        void read_next( Value_type& value )
+        {
+            posn_begin_ = read_range_or_throw( posn_begin_, posn_end_, value );
+        }
+
+    private:
+
+        typedef Multi_pass_iters< Istream_type > Mp_iters;
+        typedef spirit_namespace::position_iterator< typename Mp_iters::Mp_iter > Posn_iter_t;
+
+        Mp_iters iters_;
+        Posn_iter_t posn_begin_, posn_end_;
+    };
+}
+
+#endif

src/json_spirit_headers/json_spirit_reader.cpp

@@ -0,0 +1,63 @@
+#ifndef JSON_SPIRIT_UTILS
+#define JSON_SPIRIT_UTILS
+
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1020)
+# pragma once
+#endif
+
+#include "json_spirit_value.h"
+#include <map>
+
+namespace json_spirit
+{ 
+    template< class Obj_t, class Map_t >
+    void obj_to_map( const Obj_t& obj, Map_t& mp_obj )
+    {
+        mp_obj.clear();
+
+        for( typename Obj_t::const_iterator i = obj.begin(); i != obj.end(); ++i )
+        {
+            mp_obj[ i->name_ ] = i->value_;
+        }
+    }
+
+    template< class Obj_t, class Map_t >
+    void map_to_obj( const Map_t& mp_obj, Obj_t& obj )
+    {
+        obj.clear();
+
+        for( typename Map_t::const_iterator i = mp_obj.begin(); i != mp_obj.end(); ++i )
+        {
+            obj.push_back( typename Obj_t::value_type( i->first, i->second ) );
+        }
+    }
+
+#ifdef JSON_SPIRIT_VALUE_ENABLED
+    typedef std::map< std::string, Value > Mapped_obj;
+#endif
+
+#if defined( JSON_SPIRIT_WVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    typedef std::map< std::wstring, wValue > wMapped_obj;
+#endif
+
+    template< class Object_type, class String_type >
+    const typename Object_type::value_type::Value_type& find_value( const Object_type& obj, const String_type& name )
+    {
+        for( typename Object_type::const_iterator i = obj.begin(); i != obj.end(); ++i )
+        {
+            if( i->name_ == name )
+            {
+                return i->value_;
+            }
+        }
+
+        return Object_type::value_type::Value_type::null;
+    }
+}
+
+#endif

src/json_spirit_headers/json_spirit_reader.h

@@ -0,0 +1,8 @@
+/* Copyright (c) 2007 John W Wilkinson
+
+   This source code can be used for any purpose as long as
+   this comment is retained. */
+
+// json spirit version 2.00
+
+#include "json_spirit_value.h"

src/json_spirit_headers/json_spirit_reader_template.h

@@ -0,0 +1,96 @@
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#include "json_spirit_writer.h"
+#include "json_spirit_writer_template.h"
+
+using namespace json_spirit;
+
+#ifdef JSON_SPIRIT_VALUE_ENABLED
+    void json_spirit::write( const Value& value, std::ostream& os, unsigned int options )
+    {
+        write_stream( value, os, options );
+    }
+    std::string json_spirit::write( const Value& value, unsigned int options )
+    {
+        return write_string( value, options );
+    }
+
+    void json_spirit::write_formatted( const Value& value, std::ostream& os )
+    {
+        write_stream( value, os, pretty_print );
+    }
+
+    std::string json_spirit::write_formatted( const Value& value )
+    {
+        return write_string( value, pretty_print );
+    }
+#endif
+
+#ifdef JSON_SPIRIT_MVALUE_ENABLED
+    void json_spirit::write( const mValue& value, std::ostream& os, unsigned int options )
+    {
+        write_stream( value, os, options );
+    }
+
+    std::string json_spirit::write( const mValue& value, unsigned int options )
+    {
+        return write_string( value, options );
+    }
+
+    void json_spirit::write_formatted( const mValue& value, std::ostream& os )
+    {
+        write_stream( value, os, pretty_print );
+    }
+
+    std::string json_spirit::write_formatted( const mValue& value )
+    {
+        return write_string( value, pretty_print );
+    }
+#endif
+
+#if defined( JSON_SPIRIT_WVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    void json_spirit::write( const wValue& value, std::wostream& os, unsigned int options )
+    {
+        write_stream( value, os, options );
+    }
+
+    std::wstring json_spirit::write( const wValue& value, unsigned int options )
+    {
+        return write_string( value, options );
+    }
+
+    void json_spirit::write_formatted( const wValue& value, std::wostream& os )
+    {
+        write_stream( value, os, pretty_print );
+    }
+
+    std::wstring json_spirit::write_formatted( const wValue& value )
+    {
+        return write_string( value, pretty_print );
+    }
+#endif
+
+#if defined( JSON_SPIRIT_WMVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    void json_spirit::write_formatted( const wmValue& value, std::wostream& os )
+    {
+        write_stream( value, os, pretty_print );
+    }
+
+    std::wstring json_spirit::write_formatted( const wmValue& value )
+    {
+        return write_string( value, pretty_print );
+    }
+
+    void json_spirit::write( const wmValue& value, std::wostream& os, unsigned int options )
+    {
+        write_stream( value, os, options );
+    }
+
+    std::wstring json_spirit::write( const wmValue& value, unsigned int options )
+    {
+        return write_string( value, options );
+    }
+#endif

src/json_spirit_headers/json_spirit_stream_reader.h

@@ -0,0 +1,63 @@
+#ifndef JSON_SPIRIT_WRITER
+#define JSON_SPIRIT_WRITER
+
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1020)
+# pragma once
+#endif
+
+#include "json_spirit_value.h"
+#include "json_spirit_writer_options.h"
+#include <iostream>
+
+namespace json_spirit
+{
+    // these functions to convert JSON Values to text
+
+#ifdef JSON_SPIRIT_VALUE_ENABLED
+    void         write( const Value&  value, std::ostream&  os, unsigned int options = 0 );
+    std::string  write( const Value&  value, unsigned int options = 0 );
+#endif
+
+#ifdef JSON_SPIRIT_MVALUE_ENABLED
+    void         write( const mValue& value, std::ostream&  os, unsigned int options = 0 );
+    std::string  write( const mValue& value, unsigned int options = 0 );
+#endif
+
+#if defined( JSON_SPIRIT_WVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    void         write( const wValue&  value, std::wostream& os, unsigned int options = 0 );
+    std::wstring write( const wValue&  value, unsigned int options = 0 );
+#endif
+
+#if defined( JSON_SPIRIT_WMVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    void         write( const wmValue& value, std::wostream& os, unsigned int options = 0 );
+    std::wstring write( const wmValue& value, unsigned int options = 0 );
+#endif
+
+    // these "formatted" versions of the "write" functions are the equivalent of the above functions
+    // with option "pretty_print"
+    
+#ifdef JSON_SPIRIT_VALUE_ENABLED
+    void         write_formatted( const Value& value, std::ostream&  os );
+    std::string  write_formatted( const Value& value );
+#endif
+#ifdef JSON_SPIRIT_MVALUE_ENABLED
+    void         write_formatted( const mValue& value, std::ostream&  os );
+    std::string  write_formatted( const mValue& value );
+#endif
+
+#if defined( JSON_SPIRIT_WVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    void         write_formatted( const wValue& value, std::wostream& os );
+    std::wstring write_formatted( const wValue& value );
+#endif
+#if defined( JSON_SPIRIT_WMVALUE_ENABLED ) && !defined( BOOST_NO_STD_WSTRING )
+    void         write_formatted( const wmValue& value, std::wostream& os );
+    std::wstring write_formatted( const wmValue& value );
+#endif
+}
+
+#endif

src/json_spirit_headers/json_spirit_utils.h

@@ -0,0 +1,30 @@
+#ifndef JSON_SPIRIT_WRITER_OPTIONS
+#define JSON_SPIRIT_WRITER_OPTIONS
+
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1020)
+# pragma once
+#endif
+
+namespace json_spirit
+{
+    enum Output_options{ pretty_print = 0x01,   // Add whitespace to format the output nicely.
+
+                         raw_utf8 = 0x02,       // This prevents non-printable characters from being escapted using "\uNNNN" notation.
+                                                // Note, this is an extension to the JSON standard. It disables the escaping of
+                                                // non-printable characters allowing UTF-8 sequences held in 8 bit char strings
+                                                // to pass through unaltered.
+
+                         remove_trailing_zeros = 0x04,
+                                                // outputs e.g. "1.200000000000000" as "1.2"
+                         single_line_arrays = 0x08,
+                                                // pretty printing except that arrays printed on single lines unless they contain
+                                                // composite elements, i.e. objects or arrays
+                       };
+}
+
+#endif

src/json_spirit_headers/json_spirit_value.cpp

@@ -0,0 +1,385 @@
+#ifndef JSON_SPIRIT_WRITER_TEMPLATE
+#define JSON_SPIRIT_WRITER_TEMPLATE
+
+//          Copyright John W. Wilkinson 2007 - 2011
+// Distributed under the MIT License, see accompanying file LICENSE.txt
+
+// json spirit version 4.05
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1020)
+# pragma once
+#endif
+
+#include "json_spirit_value.h"
+#include "json_spirit_writer_options.h"
+
+#include <cassert>
+#include <sstream>
+#include <iomanip>
+#include <boost/io/ios_state.hpp>
+
+namespace json_spirit
+{
+    inline char to_hex_char( unsigned int c )
+    {
+        assert( c <= 0xF );
+
+        const char ch = static_cast< char >( c );
+
+        if( ch < 10 ) return '0' + ch;
+
+        return 'A' - 10 + ch;
+    }
+
+    template< class String_type >
+    String_type non_printable_to_string( unsigned int c )
+    {
+        typedef typename String_type::value_type Char_type;
+
+        String_type result( 6, '\\' );
+
+        result[1] = 'u';
+
+        result[ 5 ] = to_hex_char( c & 0x000F ); c >>= 4;
+        result[ 4 ] = to_hex_char( c & 0x000F ); c >>= 4;
+        result[ 3 ] = to_hex_char( c & 0x000F ); c >>= 4;
+        result[ 2 ] = to_hex_char( c & 0x000F );
+
+        return result;
+    }
+
+    template< typename Char_type, class String_type >
+    bool add_esc_char( Char_type c, String_type& s )
+    {
+        switch( c )
+        {
+            case '"':  s += to_str< String_type >( "\\\"" ); return true;
+            case '\\': s += to_str< String_type >( "\\\\" ); return true;
+            case '\b': s += to_str< String_type >( "\\b"  ); return true;
+            case '\f': s += to_str< String_type >( "\\f"  ); return true;
+            case '\n': s += to_str< String_type >( "\\n"  ); return true;
+            case '\r': s += to_str< String_type >( "\\r"  ); return true;
+            case '\t': s += to_str< String_type >( "\\t"  ); return true;
+        }
+
+        return false;
+    }
+
+    template< class String_type >
+    String_type add_esc_chars( const String_type& s, bool raw_utf8 )
+    {
+        typedef typename String_type::const_iterator Iter_type;
+        typedef typename String_type::value_type     Char_type;
+
+        String_type result;
+
+        const Iter_type end( s.end() );
+
+        for( Iter_type i = s.begin(); i != end; ++i )
+        {
+            const Char_type c( *i );
+
+            if( add_esc_char( c, result ) ) continue;
+
+            if( raw_utf8 )
+            {
+                result += c;
+            }
+            else
+            {
+                const wint_t unsigned_c( ( c >= 0 ) ? c : 256 + c );
+
+                if( iswprint( unsigned_c ) )
+                {
+                    result += c;
+                }
+                else
+                {
+                    result += non_printable_to_string< String_type >( unsigned_c );
+                }
+            }
+        }
+
+        return result;
+    }
+
+    template< class Ostream >
+    void append_double( Ostream& os, const double d, const int precision )
+    {
+        os << std::showpoint << std::setprecision( precision ) << d;
+    }
+
+    template< class String_type >
+    void erase_and_extract_exponent( String_type& str, String_type& exp )
+    {
+        const typename String_type::size_type exp_start= str.find( 'e' );
+
+        if( exp_start != String_type::npos )
+        {
+            exp = str.substr( exp_start );
+            str.erase( exp_start );
+        }
+    }
+
+    template< class String_type >
+    typename String_type::size_type find_first_non_zero( const String_type& str )
+    {
+        typename String_type::size_type result = str.size() - 1;
+
+        for( ; result != 0; --result )
+        {
+            if( str[ result ] != '0' )
+            {
+                break;
+            }
+        }
+
+        return result;
+    }
+
+    template< class String_type >
+    void remove_trailing( String_type& str )
+    {
+        String_type exp;
+
+        erase_and_extract_exponent( str, exp );
+
+        const typename String_type::size_type first_non_zero = find_first_non_zero( str );
+
+        if( first_non_zero != 0 )
+        {
+            const int offset = str[first_non_zero] == '.' ? 2 : 1;  // note zero digits following a decimal point is non standard
+            str.erase( first_non_zero + offset );
+        }
+
+        str += exp;
+    }
+
+    // this class generates the JSON text,
+    // it keeps track of the indentation level etc.
+    //
+    template< class Value_type, class Ostream_type >
+    class Generator
+    {
+        typedef typename Value_type::Config_type Config_type;
+        typedef typename Config_type::String_type String_type;
+        typedef typename Config_type::Object_type Object_type;
+        typedef typename Config_type::Array_type Array_type;
+        typedef typename String_type::value_type Char_type;
+        typedef typename Object_type::value_type Obj_member_type;
+
+    public:
+
+        Generator( const Value_type& value, Ostream_type& os, unsigned int options )
+        :   os_( os )
+        ,   indentation_level_( 0 )
+        ,   pretty_( ( options & pretty_print ) != 0 || ( options & single_line_arrays ) != 0 )
+        ,   raw_utf8_( ( options & raw_utf8 ) != 0 )
+        ,   remove_trailing_zeros_( ( options & remove_trailing_zeros ) != 0 )
+        ,   single_line_arrays_( ( options & single_line_arrays ) != 0 )
+        ,   ios_saver_( os )
+        {
+            output( value );
+        }
+
+    private:
+
+        void output( const Value_type& value )
+        {
+            switch( value.type() )
+            {
+                case obj_type:   output( value.get_obj() );   break;
+                case array_type: output( value.get_array() ); break;
+                case str_type:   output( value.get_str() );   break;
+                case bool_type:  output( value.get_bool() );  break;
+                case real_type:  output( value.get_real() );  break;
+                case int_type:   output_int( value );         break;
+                case null_type:  os_ << "null";               break;
+                default: assert( false );
+            }
+        }
+
+        void output( const Object_type& obj )
+        {
+            output_array_or_obj( obj, '{', '}' );
+        }
+
+        void output( const Obj_member_type& member )
+        {
+            output( Config_type::get_name( member ) ); space(); 
+            os_ << ':'; space(); 
+            output( Config_type::get_value( member ) );
+        }
+
+        void output_int( const Value_type& value )
+        {
+            if( value.is_uint64() )
+            {
+                os_ << value.get_uint64();
+            }
+            else
+            {
+               os_ << value.get_int64();
+            }
+        }
+
+        void output( const String_type& s )
+        {
+            os_ << '"' << add_esc_chars( s, raw_utf8_ ) << '"';
+        }
+
+        void output( bool b )
+        {
+            os_ << to_str< String_type >( b ? "true" : "false" );
+        }
+
+        void output( double d )
+        {
+            if( remove_trailing_zeros_ )
+            {
+                std::basic_ostringstream< Char_type > os;
+
+                append_double( os, d, 16 );  // note precision is 16 so that we get some trailing space that we can remove,
+                                             // otherwise, 0.1234 gets converted to "0.12399999..."
+
+                String_type str = os.str();
+
+                remove_trailing( str );
+
+                os_ << str;
+            }
+            else
+            {
+                append_double( os_, d, 17 );
+            }
+        }
+
+        static bool contains_composite_elements( const Array_type& arr )
+        {
+            for( typename Array_type::const_iterator i = arr.begin(); i != arr.end(); ++i )
+            {
+                const Value_type& val = *i;
+
+                if( val.type() == obj_type ||
+                    val.type() == array_type )
+                {
+                    return true;
+                }
+            }
+
+            return false;
+        }
+
+        template< class Iter >
+        void output_composite_item( Iter i, Iter last )
+        {
+            output( *i );
+
+            if( ++i != last )
+            {
+                os_ << ',';
+            }
+        }
+
+        void output( const Array_type& arr )
+        {
+            if( single_line_arrays_ && !contains_composite_elements( arr )  )
+            {
+                os_ << '['; space();
+               
+                for( typename Array_type::const_iterator i = arr.begin(); i != arr.end(); ++i )
+                {
+                    output_composite_item( i, arr.end() );
+
+                    space();
+                }
+
+                os_ << ']';
+            }
+            else
+            {
+                output_array_or_obj( arr, '[', ']' );
+            }
+        }
+
+        template< class T >
+        void output_array_or_obj( const T& t, Char_type start_char, Char_type end_char )
+        {
+            os_ << start_char; new_line();
+
+            ++indentation_level_;
+            
+            for( typename T::const_iterator i = t.begin(); i != t.end(); ++i )
+            {
+                indent();
+
+                output_composite_item( i, t.end() );
+
+                new_line();
+            }
+
+            --indentation_level_;
+
+            indent(); os_ << end_char;
+        }
+        
+        void indent()
+        {
+            if( !pretty_ ) return;
+
+            for( int i = 0; i < indentation_level_; ++i )
+            { 
+                os_ << "    ";
+            }
+        }
+
+        void space()
+        {
+            if( pretty_ ) os_ << ' ';
+        }
+
+        void new_line()
+        {
+            if( pretty_ ) os_ << '\n';
+        }
+
+        Generator& operator=( const Generator& ); // to prevent "assignment operator could not be generated" warning
+
+        Ostream_type& os_;
+        int indentation_level_;
+        bool pretty_;
+        bool raw_utf8_;
+        bool remove_trailing_zeros_;
+        bool single_line_arrays_;
+        boost::io::basic_ios_all_saver< Char_type > ios_saver_;  // so that ostream state is reset after control is returned to the caller
+    };
+
+    // writes JSON Value to a stream, e.g.
+    //
+    // write_stream( value, os, pretty_print );
+    //
+    template< class Value_type, class Ostream_type >
+    void write_stream( const Value_type& value, Ostream_type& os, unsigned int options = 0 )
+    {
+        os << std::dec;
+        Generator< Value_type, Ostream_type >( value, os, options );
+    }
+
+    // writes JSON Value to a stream, e.g.
+    //
+    // const string json_str = write( value, pretty_print );
+    //
+    template< class Value_type >
+    typename Value_type::String_type write_string( const Value_type& value, unsigned int options = 0 )
+    {
+        typedef typename Value_type::String_type::value_type Char_type;
+
+        std::basic_ostringstream< Char_type > os;
+
+        write_stream( value, os, options );
+
+        return os.str();
+    }
+}
+
+#endif

src/json_spirit_headers/json_spirit_value.h

@@ -0,0 +1,64 @@
+/*
+ * kvsorter_main.cc
+ *
+ *  Created on: Nov 21, 2012
+ *      Author: alex
+ *
+ * In memory key value sorter using stdin and stdout
+ */
+#include <cstdlib>
+#include <cstring>
+#include <cerrno>
+#include <stdio.h>
+#include <iostream>
+#include <string>
+#include <vector>
+#include <boost/algorithm/string.hpp>
+#include "headers/kv_struct.h"
+
+using namespace std;
+
+bool comparator(kv t1, kv t2);
+
+int main(int args, char *argv[]) {
+
+	string input_line;
+	vector<kv> data;
+	while (getline(cin, input_line)) {
+
+		//split on tab
+		vector<string> input;
+		boost::split(input, input_line, boost::is_any_of("\t"));
+
+		//get <key> <value>
+		string key = input[0];
+		string value = input[1];
+
+		//build kv struct
+		kv my_kv;
+		my_kv.key = key;
+		my_kv.value = value;
+
+		//populate data vector
+		data.push_back(my_kv);
+	}
+
+	//sort data vector with stl
+	sort(data.begin(),data.end(), comparator);
+
+	//print sorted data to stdout
+	vector<kv>::iterator it;
+	for (it = data.begin(); it != data.end(); it++) {
+		string key = (*it).key;
+		string value = (*it).value;
+
+		printf("%s\t%s\n", key.c_str(), value.c_str());
+	}
+
+	return 0;
+}
+
+bool comparator(kv t1, kv t2)
+{
+   return t1.key < t2.key ? true : false;
+}

src/json_spirit_headers/json_spirit_writer.cpp

@@ -0,0 +1,32 @@
+/*
+ * mapper_spill_checker.hpp
+ *
+ *  Created on: Nov 23, 2012
+ *      Author: alex
+ */
+
+#include "shared_mapper_data.cc"
+class mapper_spill_checker {
+private:
+	shared_mapper_data *m_mapper_data;
+
+public:
+// Constructor with id and the queue to use.
+	mapper_spill_checker(shared_mapper_data *mapper_data):m_mapper_data(mapper_data) {
+
+	}
+
+	//check mapper data
+	void operator ()() {
+		while (true) {
+
+			m_mapper_data->scan();
+
+
+
+			// Make sure we can be interrupted
+			boost::this_thread::interruption_point();
+			boost::this_thread::yield();
+		}
+	}
+};

src/json_spirit_headers/json_spirit_writer.h

@@ -0,0 +1,112 @@
+#include <cstdlib>
+#include <cstring>
+#include <cerrno>
+#include <stdio.h>
+#include <iostream>
+#include <string>
+#include <map>
+#include <list>
+#include <boost/crc.hpp>
+#include <boost/program_options.hpp>
+#include <boost/algorithm/string.hpp>
+#include <boost/thread.hpp>
+//add in_port_t type
+#include <libgearman/gearman.h>
+
+#include "kfs/KfsClient.h"
+#include "kfs/KfsAttr.h"
+
+#include "headers/kv_struct.h"
+#include "shared_mapper_data.cc"
+#include "mapper_spill_checker.hpp"
+
+using namespace std;
+KFS::KfsClient *gKfsClient;
+
+#define SPILL_THRESHOLD_BYTES 1048576 //1MB
+
+int crc32(const string& my_string);
+int get_partition(string key, int num_partitions);
+
+int main(int args, char *argv[]) {
+	in_port_t qfs_meta_server_port;
+	int  partition_count;
+	string qfs_meta_server_host, qfs_map_folder, job_id;
+
+	boost::program_options::options_description desc("Options");
+	desc.add_options()("help", "Options related to the program.")
+			("meta_server_host,s", boost::program_options::value<string>(&qfs_meta_server_host)->default_value(	"localhost"), "Connect to the qfs meta server host")
+			("meta_server_port,p", boost::program_options::value<in_port_t>(&qfs_meta_server_port)->default_value(20000), "Port number use for qfs meta server connection")
+			("qfs_map_folder,f", boost::program_options::value<string>(&qfs_map_folder)->default_value("/tmp/"), "Intermediate folder for storing mapper output")
+			("job_id", boost::program_options::value<string>(&job_id)->required(), "MR Job id")
+			("partition_count,c", boost::program_options::value<int>(&partition_count)->default_value(4), "Number of mapped output partitions");
+
+	boost::program_options::variables_map vm;
+	try {
+		boost::program_options::store(
+				boost::program_options::parse_command_line(args, argv, desc),
+				vm);
+		boost::program_options::notify(vm);
+	} catch (exception &e) {
+		cout << e.what() << endl;
+		return EXIT_FAILURE;
+	}
+
+	if (vm.count("help")) {
+		cout << desc << endl;
+		return EXIT_SUCCESS;
+	}
+
+	//init
+	string input_line;
+	shared_mapper_data mapper_data(qfs_meta_server_host, qfs_meta_server_port, qfs_map_folder, job_id, partition_count, SPILL_THRESHOLD_BYTES);
+
+	//start mapper_data spill thread
+	mapper_spill_checker c(&mapper_data);
+	boost::thread t(c);
+
+	while (getline(cin, input_line)) {
+
+		//split on tab
+		vector<string> input;
+		boost::split(input, input_line, boost::is_any_of("\t"));
+
+		//get <key> <value>
+		string key = input[0];
+		string value = input[1];
+
+		//build kv struct
+		kv my_kv;
+		my_kv.key = key;
+		my_kv.value = value;
+
+		//partition
+		int partition = get_partition(my_kv.key, partition_count);
+
+		//save
+		mapper_data.add(partition, my_kv);
+
+	}
+
+	//flush any remaining mapper_data
+	mapper_data.flush_all();
+
+	//wait for thread
+	t.interrupt();
+	t.join();
+
+	return 0;
+}
+
+//crc calculation
+int crc32(const string& my_string) {
+    boost::crc_32_type result;
+    result.process_bytes(my_string.data(), my_string.length());
+    return result.checksum();
+}
+
+//simple random hash partitioning
+int get_partition(string key, int num_partitions){
+	return abs(crc32(key)) % num_partitions;
+}
+

src/json_spirit_headers/json_spirit_writer_options.h

@@ -0,0 +1,189 @@
+/*
+ ============================================================================
+ Name        : mapper_worker_main.cpp
+ Author      : Alexander Hurd
+ Version     :
+ Copyright   :
+ Description :
+ ============================================================================
+ */
+
+#include <cerrno>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+
+#include <libgearman/gearman.h>
+#include <boost/program_options.hpp>
+#include <boost/lexical_cast.hpp>
+
+#include "json_spirit_headers/json_spirit_reader_template.h"
+#include "headers/workloads.h"
+
+#ifndef __INTEL_COMPILER
+#pragma GCC diagnostic ignored "-Wold-style-cast"
+#endif
+
+using namespace std;
+using namespace json_spirit;
+
+static void *mapper(gearman_job_st *job, void *context, size_t *result_size,
+		gearman_return_t *ret_ptr);
+
+string path_to_qfs_bin_tools, path_to_qfs_mapred_bin;
+int main(int args, char *argv[]) {
+	uint32_t count;
+	int timeout;
+	in_port_t port;
+	string host;
+
+	boost::program_options::options_description desc("Options");
+	desc.add_options()("help", "Options related to the program.")("host,h",
+			boost::program_options::value<string>(&host)->default_value(
+					"localhost"), "Connect to the gearmand host")("port,p",
+			boost::program_options::value<in_port_t>(&port)->default_value(
+					GEARMAN_DEFAULT_TCP_PORT),
+			"Port number use for gearmand connection")("count,c",
+			boost::program_options::value<uint32_t>(&count)->default_value(0),
+			"Number of jobs to run before exiting")("timeout,u",
+			boost::program_options::value<int>(&timeout)->default_value(-1),
+			"Timeout in milliseconds")("path_to_qfs_bin_tools",
+			boost::program_options::value<string>(&path_to_qfs_bin_tools)->default_value(
+					"/home/alex/qfs/build/release/bin/tools/"),
+			"Path to qfs tools folder")("path_to_qfs_mapred_bin",
+			boost::program_options::value<string>(&path_to_qfs_mapred_bin)->default_value(
+					"/home/alex/workspace/qfs_mapred/src/"),
+			"Path to qfs_mapred bin folder");
+
+	boost::program_options::variables_map vm;
+	try {
+		boost::program_options::store(
+				boost::program_options::parse_command_line(args, argv, desc),
+				vm);
+		boost::program_options::notify(vm);
+	} catch (exception &e) {
+		cout << e.what() << endl;
+		return EXIT_FAILURE;
+	}
+
+	if (vm.count("help")) {
+		cout << desc << endl;
+		return EXIT_SUCCESS;
+	}
+
+	gearman_worker_st worker;
+	if (gearman_worker_create(&worker) == NULL) {
+		cerr << "Memory allocation failure on worker creation." << endl;
+		return EXIT_FAILURE;
+	}
+
+	if (timeout >= 0)
+		gearman_worker_set_timeout(&worker, timeout);
+
+	gearman_return_t ret;
+	ret = gearman_worker_add_server(&worker, host.c_str(), port);
+	if (ret != GEARMAN_SUCCESS) {
+		cerr << gearman_worker_error(&worker) << endl;
+		return EXIT_FAILURE;
+	}
+
+	ret = gearman_worker_add_function(&worker, "mapper", 0, mapper, NULL);
+	if (ret != GEARMAN_SUCCESS) {
+		cerr << gearman_worker_error(&worker) << endl;
+		return EXIT_FAILURE;
+	}
+
+	while (1) {
+		ret = gearman_worker_work(&worker);
+		if (ret != GEARMAN_SUCCESS) {
+			cerr << gearman_worker_error(&worker) << endl;
+			break;
+		}
+
+		if (count > 0) {
+			count--;
+			if (count == 0)
+				break;
+		}
+	}
+
+	gearman_worker_free(&worker);
+
+	return EXIT_SUCCESS;
+}
+
+static void *mapper(gearman_job_st *job, void *context, size_t *result_size,
+		gearman_return_t *ret_ptr) {
+
+	const char *workload;
+	workload = (const char *) gearman_job_workload(job);
+
+	//build payload
+	string json_workload = (string) workload;
+	Value value;
+	read_string(json_workload, value);
+	Object obj = value.get_obj();
+	Mapper_Workload payload;
+	for (Object::size_type i = 0; i != obj.size(); ++i) {
+		const Pair& pair = obj[i];
+
+		const string& name = pair.name_;
+		const Value& value = pair.value_;
+
+		if (name == "job_id") {
+			payload.job_id = value.get_str();
+		} else if (name == "qfs_meta_server_name") {
+			payload.qfs_meta_server_name = value.get_str();
+		} else if (name == "qfs_meta_server_port") {
+			payload.qfs_meta_server_port = value.get_int();
+		} else if (name == "qfs_file_input") {
+			payload.qfs_file_input = value.get_str();
+		} else if (name == "qfs_map_folder") {
+			payload.qfs_map_folder = value.get_str();
+		} else if (name == "python_mapper_function") {
+			payload.python_mapper_function = value.get_str();
+		} else if (name == "partition_count") {
+			payload.partition_count = value.get_int();
+		} else {
+			assert( false);
+		}
+	}
+
+	//write maper function to file disk
+	const char *python_mapper_file = tmpnam(NULL);  // Get temp name
+	FILE *fp = fopen(python_mapper_file, "w");  // Create the file
+	fwrite(payload.python_mapper_function.c_str(), 1,
+			payload.python_mapper_function.size(), fp);
+	fclose(fp);
+
+	printf("Checking if processor is available...");
+	if (!system(NULL)) {
+		// *ret_ptr = GEARMAN_ERROR;
+		// return;
+	}
+
+	int r;
+	string mapper_command = path_to_qfs_bin_tools + "cpfromqfs -s "
+			+ payload.qfs_meta_server_name + " -p "
+			+ boost::lexical_cast<string>(payload.qfs_meta_server_port) + " -k "
+			+ payload.qfs_file_input + " -d - | python " + python_mapper_file
+			+ " | " + path_to_qfs_mapred_bin + "mapper_to_qfs_partitions -s " + payload.qfs_meta_server_name
+			+ " -p " + boost::lexical_cast<string>(payload.qfs_meta_server_port)
+			+ " -f " + payload.qfs_map_folder + " --job_id " + payload.job_id
+			+ " --partition_count "
+			+ boost::lexical_cast<string>(payload.partition_count);
+	r = system(mapper_command.c_str());
+
+	//output results if any
+	cout << r;
+
+	//remove python script
+	unlink(python_mapper_file);
+
+	*ret_ptr = GEARMAN_SUCCESS;
+
+	return NULL;
+
+}
+

src/json_spirit_headers/json_spirit_writer_template.h

@@ -0,0 +1,103 @@
+/*
+ * qfs_atomic_writer.hpp
+ *
+ *  Created on: Nov 23, 2012
+ *      Author: alex
+ */
+#include <map>
+#include <fstream>
+#include <fcntl.h>
+#include "kfs/KfsClient.h"
+#include "kfs/KfsAttr.h"
+#include "kfs/common/kfstypes.h"
+#include "shared_queue.hpp"
+#include "headers/mapper_data_struct.h"
+
+using namespace std;
+
+// Class that consumes objects from a queue
+class qfs_atomic_writer {
+private:
+	shared_queue<mapper_data>* m_queue; // The queue to use
+	string m_qfs_map_folder;
+	string m_job_id;
+	map<int, int> m_qfs_partition_fd;
+	KFS::KfsClient *m_KfsClient;
+
+public:
+	qfs_atomic_writer(shared_queue<mapper_data>* queue,
+			string qfs_meta_server_host, in_port_t qfs_meta_server_port,
+			string qfs_map_folder, string job_id) :
+			m_queue(queue), m_qfs_map_folder(qfs_map_folder), m_job_id(job_id) {
+
+		//connect to qfs
+		m_KfsClient = KFS::Connect(qfs_meta_server_host, qfs_meta_server_port);
+		if (!m_KfsClient) {
+			cerr << "kfs client failed to initialize...exiting" << "\n";
+			exit(-1);
+		}
+	}
+
+	~qfs_atomic_writer() {
+
+		//close partitions fds
+		map<int, int>::iterator it;
+		for (it = m_qfs_partition_fd.begin(); it != m_qfs_partition_fd.end();
+				it++) {
+			int fd = (*it).second;
+			m_KfsClient->Close(fd);
+		}
+	}
+
+	// The thread function reads data from the queue
+	void operator ()() {
+		while (true) {
+
+			// Get the data from the queue
+			mapper_data data = m_queue->Dequeue();
+
+			string my_data = data.data;
+			int partition = data.partition;
+
+			//check for fd
+			int fd;
+			if (m_qfs_partition_fd.find(partition)
+					== m_qfs_partition_fd.end()) {
+				//init fd
+				init_qfs_fd(partition);
+			}
+
+			//get partition fd
+			fd = m_qfs_partition_fd[partition];
+
+			//write!
+			int res = m_KfsClient->AtomicRecordAppend(fd, my_data.c_str(), my_data.size());
+			if (res != (int) my_data.size())
+				cout << "Atomic write err" <<endl;
+
+			// Make sure we can be interrupted
+			boost::this_thread::interruption_point();
+			boost::this_thread::yield();
+
+		}
+	}
+
+	void init_qfs_fd(int partition) {
+		string mapper_partition_file = m_qfs_map_folder
+				+ m_job_id + "_"
+				+ boost::lexical_cast<string>(partition);
+
+		int fd;
+		if ((fd = m_KfsClient->Open(mapper_partition_file.c_str(), O_APPEND))
+				< 0) {
+			cout << "Mapper Partition File Open on : " << mapper_partition_file
+					<< " failed: " << KFS::ErrorCodeToStr(fd) << endl;
+			exit(-1);
+		}
+
+		//save
+		m_qfs_partition_fd[partition] = fd;
+
+	}
+};
+

src/kvsorter_main.cc

@@ -0,0 +1,187 @@
+/*
+ ============================================================================
+ Name        : reducer_worker.cpp
+ Author      : Alexander Hurd
+ Version     :
+ Copyright   :
+ Description :
+ ============================================================================
+ */
+
+#include <cerrno>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+
+#include <libgearman/gearman.h>
+#include <boost/program_options.hpp>
+#include <boost/lexical_cast.hpp>
+
+#include "json_spirit_headers/json_spirit_reader_template.h"
+#include "headers/workloads.h"
+
+#ifndef __INTEL_COMPILER
+#pragma GCC diagnostic ignored "-Wold-style-cast"
+#endif
+
+using namespace std;
+using namespace json_spirit;
+
+static void *reducer(gearman_job_st *job, void *context, size_t *result_size,
+		gearman_return_t *ret_ptr);
+
+string path_to_qfs_bin_tools;
+int main(int args, char *argv[]) {
+	uint32_t count;
+	int timeout;
+	in_port_t port;
+	string host;
+
+	boost::program_options::options_description desc("Options");
+	desc.add_options()("help", "Options related to the program.")("host,h",
+			boost::program_options::value<string>(&host)->default_value(
+					"localhost"), "Connect to the gearmand host")("port,p",
+			boost::program_options::value<in_port_t>(&port)->default_value(
+					GEARMAN_DEFAULT_TCP_PORT),
+			"Port number use for gearmand connection")("count,c",
+			boost::program_options::value<uint32_t>(&count)->default_value(0),
+			"Number of jobs to run before exiting")("timeout,u",
+			boost::program_options::value<int>(&timeout)->default_value(-1),
+			"Timeout in milliseconds")("path_to_qfs_bin_tools,p",
+			boost::program_options::value<string>(&path_to_qfs_bin_tools)->default_value(
+					"/home/alex/qfs/build/release/bin/tools/"),
+			"Path to qfs tools folder");
+	;
+
+	boost::program_options::variables_map vm;
+	try {
+		boost::program_options::store(
+				boost::program_options::parse_command_line(args, argv, desc),
+				vm);
+		boost::program_options::notify(vm);
+	} catch (exception &e) {
+		cout << e.what() << endl;
+		return EXIT_FAILURE;
+	}
+
+	if (vm.count("help")) {
+		cout << desc << endl;
+		return EXIT_SUCCESS;
+	}
+
+	gearman_worker_st worker;
+	if (gearman_worker_create(&worker) == NULL) {
+		cerr << "Memory allocation failure on worker creation." << endl;
+		return EXIT_FAILURE;
+	}
+
+	if (timeout >= 0)
+		gearman_worker_set_timeout(&worker, timeout);
+
+	gearman_return_t ret;
+	ret = gearman_worker_add_server(&worker, host.c_str(), port);
+	if (ret != GEARMAN_SUCCESS) {
+		cerr << gearman_worker_error(&worker) << endl;
+		return EXIT_FAILURE;
+	}
+
+	ret = gearman_worker_add_function(&worker, "reducer", 0, reducer, NULL);
+	if (ret != GEARMAN_SUCCESS) {
+		cerr << gearman_worker_error(&worker) << endl;
+		return EXIT_FAILURE;
+	}
+
+	while (1) {
+		ret = gearman_worker_work(&worker);
+		if (ret != GEARMAN_SUCCESS) {
+			cerr << gearman_worker_error(&worker) << endl;
+			break;
+		}
+
+		if (count > 0) {
+			count--;
+			if (count == 0)
+				break;
+		}
+	}
+
+	gearman_worker_free(&worker);
+
+	return EXIT_SUCCESS;
+}
+
+static void *reducer(gearman_job_st *job, void *context, size_t *result_size,
+		gearman_return_t *ret_ptr) {
+
+	const char *workload;
+	workload = (const char *) gearman_job_workload(job);
+
+	//build payload
+	string json_workload = (string) workload;
+	Value value;
+	read_string(json_workload, value);
+	Object obj = value.get_obj();
+	Reducer_Workload payload;
+	for (Object::size_type i = 0; i != obj.size(); ++i) {
+		const Pair& pair = obj[i];
+
+		const string& name = pair.name_;
+		const Value& value = pair.value_;
+
+		if (name == "qfs_meta_server_name") {
+			payload.qfs_meta_server_name = value.get_str();
+		} else if (name == "qfs_meta_server_port") {
+			payload.qfs_meta_server_port = value.get_int();
+		} else if (name == "qfs_sorted_partition_file") {
+			payload.qfs_sorted_partition_files = value.get_array();
+		} else if (name == "qfs_output_file") {
+			payload.qfs_output_file = value.get_str();
+		} else if (name == "python_reducer_function") {
+			payload.python_reducer_function = value.get_str();
+		} else {
+			assert( false);
+		}
+	}
+
+	//write reducer function to file disk
+	const char *python_reducer_file = tmpnam(NULL);  // Get temp name
+	FILE *fp = fopen(python_reducer_file, "w");  // Create the file
+	fwrite(payload.python_reducer_function.c_str(), 1,
+			payload.python_reducer_function.size(), fp);
+	fclose(fp);
+
+	printf("Checking if processor is available...");
+	if (!system(NULL)) {
+		// *ret_ptr = GEARMAN_ERROR;
+		// return;
+	}
+
+	//loop sorted partitions and reduce!
+	for (int i = 0; i < payload.qfs_sorted_partition_files.size(); i++) {
+
+		string file = payload.qfs_sorted_partition_files[i].get_str();
+
+		int r;
+		string mapper_command = path_to_qfs_bin_tools + "cpfromqfs -s "
+				+ payload.qfs_meta_server_name + " -p "
+				+ boost::lexical_cast<string>(payload.qfs_meta_server_port)
+				+ " -k " + file
+				+ " -d - | python " + python_reducer_file + " | " + path_to_qfs_bin_tools + "cptoqfs -s "
+				+ payload.qfs_meta_server_name + " -p "
+				+ boost::lexical_cast<string>(payload.qfs_meta_server_port)
+				+ " -d - -k " + payload.qfs_output_file + "_" + boost::lexical_cast<string>(i);
+
+		r = system(mapper_command.c_str());
+
+		//output results if any
+		cout << r;
+	}
+
+	//remove python script
+	unlink(python_reducer_file);
+
+	*ret_ptr = GEARMAN_SUCCESS;
+
+	return NULL;
+}

src/mapper_spill_checker.hpp

@@ -0,0 +1,141 @@
+/*
+ * shared_mapper_data.cc
+ *
+ *  Created on: Nov 22, 2012
+ *      Author: alex
+ */
+
+#ifndef SHARED_MAPPER_DATA_H_
+#define SHARED_MAPPER_DATA_H_
+
+#include <map>
+#include <list>
+#include <boost/thread.hpp>
+
+#include "shared_queue.hpp"
+#include "qfs_atomic_writer.hpp"
+#include "headers/mapper_data_struct.h"
+#include "headers/kv_struct.h"
+
+#define QFS_WRITER_THREADS 1
+
+using namespace std;
+
+class shared_mapper_data {
+
+private:
+	boost::thread_group m_writers;
+	map<int, list<kv> > m_mapper_data;
+	map<int, long> m_partition_sizes;
+	int m_partition_count;
+	long m_spill_threshold_kb;
+	shared_queue<mapper_data> m_write_queue;
+	boost::mutex m_mutex;
+
+public:
+	shared_mapper_data(string qfs_meta_server_host, in_port_t qfs_meta_server_port, string qfs_map_folder, string job_id, int partition_count,
+			long spill_threshold_kb) :
+			m_partition_count(partition_count), m_spill_threshold_kb(
+					spill_threshold_kb) {
+
+		// Create qfs writers
+		for (int i = 0; i < QFS_WRITER_THREADS; i++) {
+			qfs_atomic_writer w(&m_write_queue, qfs_meta_server_host, qfs_meta_server_port, qfs_map_folder, job_id);
+			m_writers.create_thread(w);
+		}
+
+	}
+
+	~shared_mapper_data() {
+		m_writers.interrupt_all();
+		m_writers.join_all();
+	}
+
+	//scan mapper partition data for spill threshold
+	void scan() {
+
+		//get lock
+		boost::unique_lock<boost::mutex> lock(m_mutex);
+
+		//scan partition size
+		map<int, long>::iterator it;
+		for (it = m_partition_sizes.begin(); it != m_partition_sizes.end();
+				it++) {
+			int partition = (*it).first;
+			long size_bytes = (*it).second;
+
+			//spill to qfs if threshold is met
+			if (size_bytes >= m_spill_threshold_kb){
+				write_partition_to_qfs(partition);
+				m_partition_sizes[partition] = 0;
+			}
+		}
+
+		//release
+		//lock.release();
+	}
+
+	//spill to qfs
+	void write_partition_to_qfs(int partition) {
+
+		//get data from map
+		list<kv> my_list = m_mapper_data[partition];
+		string data;
+		list<kv>::iterator it;
+		for (it = my_list.begin(); it != my_list.end(); it++) {
+			string key = (*it).key;
+			string value = (*it).value;
+			data.append(key + "\t" + value + "\n");
+		}
+
+		//empty list
+		m_mapper_data[partition].clear();
+
+		mapper_data my_data;
+		my_data.data = data;
+		my_data.partition = partition;
+
+		//add to write queue
+		m_write_queue.Enqueue(my_data);
+
+	}
+
+	//add new data
+	void add(int partition, kv my_kv) {
+
+		//get lock
+		boost::unique_lock<boost::mutex> lock(m_mutex);
+
+		//add kv to partition
+		m_mapper_data[partition].push_back(my_kv);
+
+		//increment partition size
+		if (m_partition_sizes.find(partition) != m_partition_sizes.end()) {
+			m_partition_sizes[partition] += my_kv.key.size()
+					+ my_kv.value.size();
+		} else {
+			//init
+			m_partition_sizes[partition] = my_kv.key.size()
+					+ my_kv.value.size();
+		}
+	}
+
+	//flush all data to qfs
+	void flush_all() {
+
+		//get lock
+		boost::unique_lock<boost::mutex> lock(m_mutex);
+
+		//loop partitions
+		map<int, list<kv> >::iterator it;
+		for (it = m_mapper_data.begin(); it != m_mapper_data.end(); it++) {
+			int partition = (*it).first;
+			list<kv> map_list = (*it).second;
+
+			//atomic append list to partition file
+			write_partition_to_qfs(partition);
+		}
+	}
+};
+
+#endif /*SHARED_MAPPER_DATA_H_*/

src/mapper_to_qfs_partitions_main.cc

@@ -0,0 +1,57 @@
+/*
+ * shared_queue.hpp
+ *
+ *  Created on: Nov 23, 2012
+ *      Author: https://www.quantnet.com/threads/c-multithreading-in-boost.10028/
+ */
+
+#ifndef SHARED_QUEUE_H_
+#define SHARED_QUEUE_H_
+
+#include <queue>
+#include <boost/thread.hpp>
+
+// Queue class that has thread synchronisation
+template<typename T>
+class shared_queue {
+private:
+	std::queue<T> m_queue; // Use STL queue to store data
+	boost::mutex m_mutex; // The mutex to synchronise on
+	boost::condition_variable m_cond; // The condition to wait for
+
+public:
+
+	// Add data to the queue and notify others
+	void Enqueue(const T& data) {
+		// Acquire lock on the queue
+		boost::unique_lock<boost::mutex> lock(m_mutex);
+
+		// Add the data to the queue
+		m_queue.push(data);
+
+		// Notify others that data is ready
+		m_cond.notify_one();
+
+	} // Lock is automatically released here
+
+	// Get data from the queue. Wait for data if not available
+	T Dequeue() {
+
+		// Acquire lock on the queue
+		boost::unique_lock<boost::mutex> lock(m_mutex);
+
+		// When there is no data, wait till someone fills it.
+		// Lock is automatically released in the wait and obtained
+		// again after the wait
+		while (m_queue.size() == 0)
+			m_cond.wait(lock);
+
+		// Retrieve the data from the queue
+		T result = m_queue.front();
+		m_queue.pop();
+		return result;
+
+	} // Lock is automatically released here
+};
+
+#endif /*SHARED_QUEUE_H_*/

src/mapper_worker_main.cc

@@ -0,0 +1,169 @@
+/*
+ * sorter_worker_main.cc
+ *
+ *  Created on: Nov 21, 2012
+ *      Author: alex
+ */
+
+#include <cerrno>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+
+#include <libgearman/gearman.h>
+#include <boost/program_options.hpp>
+#include <boost/lexical_cast.hpp>
+
+#include "json_spirit_headers/json_spirit_reader_template.h"
+#include "headers/workloads.h"
+
+#ifndef __INTEL_COMPILER
+#pragma GCC diagnostic ignored "-Wold-style-cast"
+#endif
+
+using namespace std;
+using namespace json_spirit;
+
+static void *sorter(gearman_job_st *job, void *context, size_t *result_size,
+		gearman_return_t *ret_ptr);
+
+string path_to_qfs_bin_tools, path_to_qfs_mapred_bin;
+int main(int args, char *argv[]) {
+	uint32_t count;
+	int timeout;
+	in_port_t port;
+	string host;
+
+	boost::program_options::options_description desc("Options");
+	desc.add_options()("help", "Options related to the program.")("host,h",
+			boost::program_options::value<string>(&host)->default_value(
+					"localhost"), "Connect to the gearmand host")("port,p",
+			boost::program_options::value<in_port_t>(&port)->default_value(
+					GEARMAN_DEFAULT_TCP_PORT),
+			"Port number use for gearmand connection")("count,c",
+			boost::program_options::value<uint32_t>(&count)->default_value(0),
+			"Number of jobs to run before exiting")("timeout,u",
+			boost::program_options::value<int>(&timeout)->default_value(-1),
+			"Timeout in milliseconds")("path_to_qfs_bin_tools,p",
+			boost::program_options::value<string>(&path_to_qfs_bin_tools)->default_value(
+					"/home/alex/qfs/build/release/bin/tools/"),
+			"Path to qfs tools folder")("path_to_qfs_mapred_bin",
+			boost::program_options::value<string>(&path_to_qfs_mapred_bin)->default_value(
+					"/home/alex/workspace/qfs_mapred/src/"),
+			"Path to qfs_mapred bin folder");
+
+	boost::program_options::variables_map vm;
+	try {
+		boost::program_options::store(
+				boost::program_options::parse_command_line(args, argv, desc),
+				vm);
+		boost::program_options::notify(vm);
+	} catch (exception &e) {
+		cout << e.what() << endl;
+		return EXIT_FAILURE;
+	}
+
+	if (vm.count("help")) {
+		cout << desc << endl;
+		return EXIT_SUCCESS;
+	}
+
+	gearman_worker_st worker;
+	if (gearman_worker_create(&worker) == NULL) {
+		cerr << "Memory allocation failure on worker creation." << endl;
+		return EXIT_FAILURE;
+	}
+
+	if (timeout >= 0)
+		gearman_worker_set_timeout(&worker, timeout);
+
+	gearman_return_t ret;
+	ret = gearman_worker_add_server(&worker, host.c_str(), port);
+	if (ret != GEARMAN_SUCCESS) {
+		cerr << gearman_worker_error(&worker) << endl;
+		return EXIT_FAILURE;
+	}
+
+	ret = gearman_worker_add_function(&worker, "sorter", 0, sorter, NULL);
+	if (ret != GEARMAN_SUCCESS) {
+		cerr << gearman_worker_error(&worker) << endl;
+		return EXIT_FAILURE;
+	}
+
+	while (1) {
+		ret = gearman_worker_work(&worker);
+		if (ret != GEARMAN_SUCCESS) {
+			cerr << gearman_worker_error(&worker) << endl;
+			break;
+		}
+
+		if (count > 0) {
+			count--;
+			if (count == 0)
+				break;
+		}
+	}
+
+	gearman_worker_free(&worker);
+
+	return EXIT_SUCCESS;
+}
+
+static void *sorter(gearman_job_st *job, void *context, size_t *result_size,
+		gearman_return_t *ret_ptr) {
+
+	const char *workload;
+	workload = (const char *) gearman_job_workload(job);
+
+	//build payload
+	string json_workload = (string) workload;
+	Value value;
+	read_string(json_workload, value);
+	Object obj = value.get_obj();
+	Sorter_Workload payload;
+	for (Object::size_type i = 0; i != obj.size(); ++i) {
+		const Pair& pair = obj[i];
+
+		const string& name = pair.name_;
+		const Value& value = pair.value_;
+
+		if (name == "qfs_meta_server_name") {
+			payload.qfs_meta_server_name = value.get_str();
+		} else if (name == "qfs_meta_server_port") {
+			payload.qfs_meta_server_port = value.get_int();
+		} else if (name == "qfs_partition_file_input") {
+			payload.qfs_partition_file_input = value.get_str();
+		} else if (name == "qfs_partition_file_output") {
+			payload.qfs_partition_file_output = value.get_str();
+		} else {
+			assert( false);
+		}
+	}
+
+	printf("Checking if processor is available...");
+	if (!system(NULL)) {
+		// *ret_ptr = GEARMAN_ERROR;
+		// return;
+	}
+
+	int r;
+	string sort_command = path_to_qfs_bin_tools + "cpfromqfs -s "
+			+ payload.qfs_meta_server_name + " -p "
+			+ boost::lexical_cast<string>(payload.qfs_meta_server_port) + " -k "
+			+ payload.qfs_partition_file_input
+			+ " -d - | " + path_to_qfs_mapred_bin + "kvsorter | " + path_to_qfs_bin_tools + "cptoqfs -s " + payload.qfs_meta_server_name
+			+ " -p " + boost::lexical_cast<string>(payload.qfs_meta_server_port)
+			+ " -d - -k " + payload.qfs_partition_file_output + " -S";
+
+	r = system(sort_command.c_str());
+
+	//output results if any
+	cout << r;
+
+	*ret_ptr = GEARMAN_SUCCESS;
+
+	return NULL;
+
+}
+