diff --git a/simplemr/.api b/simplemr/.api
new file mode 100644
index 0000000..16b6f34
--- /dev/null
+++ b/simplemr/.api
@@ -0,0 +1,17 @@
+pkg simplemr, method (*Identity) Map(*MR, string, interface{}) error
+pkg simplemr, method (*Identity) Reduce(*MR, string, []interface{}) error
+pkg simplemr, method (*MR) Error() error
+pkg simplemr, method (*MR) MapOut(string, ...interface{})
+pkg simplemr, method (*MR) ReduceOut(string, ...interface{})
+pkg simplemr, method (*MR) Run(<-chan *Record, chan<- *Record, Mapper, Reducer) error
+pkg simplemr, type Identity struct
+pkg simplemr, type MR struct
+pkg simplemr, type MR struct, NumMappers int
+pkg simplemr, type MR struct, Timeout time.Duration
+pkg simplemr, type Mapper interface { Map }
+pkg simplemr, type Mapper interface, Map(*MR, string, interface{}) error
+pkg simplemr, type Record struct
+pkg simplemr, type Record struct, Key string
+pkg simplemr, type Record struct, Values []interface{}
+pkg simplemr, type Reducer interface { Reduce }
+pkg simplemr, type Reducer interface, Reduce(*MR, string, []interface{}) error
diff --git a/simplemr/example_test.go b/simplemr/example_test.go
new file mode 100644
index 0000000..c3df40a
--- /dev/null
+++ b/simplemr/example_test.go
@@ -0,0 +1,31 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package simplemr_test
+
+import (
+	"fmt"
+
+	"v.io/x/lib/simplemr"
+)
+
+func ExampleMR() {
+	in, out := make(chan *simplemr.Record, 2), make(chan *simplemr.Record, 2)
+	mr := &simplemr.MR{}
+	identity := &simplemr.Identity{}
+	go mr.Run(in, out, identity, identity)
+	in <- &simplemr.Record{"1", []interface{}{"hello\n"}}
+	in <- &simplemr.Record{"2", []interface{}{"world\n"}}
+	close(in)
+	k := <-out
+	fmt.Printf("%s: %s", k.Key, k.Values[0].(string))
+	k = <-out
+	fmt.Printf("%s: %s", k.Key, k.Values[0].(string))
+	if err := mr.Error(); err != nil {
+		fmt.Printf("mr failed: %v", err)
+	}
+	// Output:
+	// 1: hello
+	// 2: world
+}
diff --git a/simplemr/mr.go b/simplemr/mr.go
new file mode 100644
index 0000000..09d0347
--- /dev/null
+++ b/simplemr/mr.go
@@ -0,0 +1,192 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package simplemr provides a simple map reduce framework for use by
+// commandline and other tools and consequently can only be used from
+// within a single process. It is specifically not intended to support
+// large datasets, but mappers are run concurrently so that long running
+// tasks (e.g. external shell commands will be run in parallel). The
+// current implementation supoorts only a single reducer however future
+// implementations are likely to run multiple reducers and hence reducers
+// should be coded accordingly.
+package simplemr
+
+import (
+	"fmt"
+	"runtime"
+	"sort"
+	"sync"
+	"time"
+)
+
+// Mapper is in the interface that must be implemented by all mappers.
+type Mapper interface {
+	// Map is called by the framework for every key, value pair read
+	// from the specified input.
+	Map(mr *MR, key string, value interface{}) error
+}
+
+// Reducer is the interface that must be implemented by the reducer.
+type Reducer interface {
+	// Reduce is called by the framework for every key and associated
+	// values that are emitted by the Mappers.
+	Reduce(mr *MR, key string, values []interface{}) error
+}
+
+// Record represents all input and output data.
+type Record struct {
+	Key    string
+	Values []interface{}
+}
+
+type store struct {
+	sync.Mutex
+	data map[string][]interface{}
+}
+
+func newStore() *store {
+	return &store{data: make(map[string][]interface{})}
+}
+
+func (s *store) sortedKeys() []string {
+	s.Lock()
+	defer s.Unlock()
+	keys := make([]string, 0, len(s.data))
+	for k, _ := range s.data {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	return keys
+}
+
+func (s *store) insert(k string, v ...interface{}) {
+	s.Lock()
+	defer s.Unlock()
+	s.data[k] = append(s.data[k], v...)
+}
+
+func (s *store) lookup(k string) []interface{} {
+	s.Lock()
+	defer s.Unlock()
+	return s.data[k]
+}
+
+// MR represents the Map Reduction.
+type MR struct {
+	input  <-chan *Record
+	output chan<- *Record
+	err    error
+	data   *store
+	// The number of conccurent mappers to use. A value of 0 instructs
+	// the implementation to use an appropriate number, such as the number
+	// of available CPUs.
+	NumMappers int
+	// The time to wait for the map reduce to complete. A value of 0 implies
+	// no timeout - i.e. an infinite wait.
+	Timeout time.Duration
+}
+
+// Error returns any error that was returned by the Run method. It is
+// safe to read its value once the output channel passed to Run has been
+// closed.
+func (mr *MR) Error() error {
+	return mr.err
+}
+
+// MapOut outputs the key and associated values for subsequent
+// processing by a Reducer. It should only be called from a mapper.
+func (mr *MR) MapOut(key string, values ...interface{}) {
+	mr.data.insert(key, values...)
+}
+
+// ReduceOut outputs the key and associated values to the specified output
+// stream. It should only be called from a reducer.
+func (mr *MR) ReduceOut(key string, values ...interface{}) {
+	mr.output <- &Record{key, values}
+}
+
+func (mr *MR) runMapper(ch chan error, mapper Mapper) {
+	for {
+		rec := <-mr.input
+		if rec == nil {
+			ch <- nil
+			return
+		}
+		for _, v := range rec.Values {
+			if err := mapper.Map(mr, rec.Key, v); err != nil {
+				ch <- err
+				return
+			}
+		}
+	}
+}
+
+func (mr *MR) runMappers(mapper Mapper, timeout <-chan time.Time) error {
+	ch := make(chan error, mr.NumMappers)
+	for i := 0; i < mr.NumMappers; i++ {
+		go mr.runMapper(ch, mapper)
+	}
+	done := 0
+	for {
+		select {
+		case err := <-ch:
+			if err != nil {
+				// We should probably drain the channel.
+				return err
+			}
+			done++
+			if done == mr.NumMappers {
+				return nil
+			}
+		case <-timeout:
+			return fmt.Errorf("timed out mappers after %s", mr.Timeout)
+		}
+	}
+}
+
+func (mr *MR) runReducers(reducer Reducer, timeout <-chan time.Time) error {
+	ch := make(chan error, 1)
+	go func() {
+		for _, k := range mr.data.sortedKeys() {
+			v := mr.data.lookup(k)
+			if err := reducer.Reduce(mr, k, v); err != nil {
+				ch <- err
+			}
+		}
+		close(ch)
+	}()
+	var err error
+	select {
+	case err = <-ch:
+	case <-timeout:
+		err = fmt.Errorf("timed out reducers after %s", mr.Timeout)
+	}
+	return err
+}
+
+// Run runs the map reduction using the supplied mapper and reducer reading
+// from input and writing to output. The caller must close the input channel
+// when there is no more input data. The implementation of Run will close
+// the output channel when the Reducer has processed all intermediate data.
+// Run may only be called once per MR receiver.
+func (mr *MR) Run(input <-chan *Record, output chan<- *Record, mapper Mapper, reducer Reducer) error {
+	mr.input, mr.output, mr.data = input, output, newStore()
+	if mr.NumMappers == 0 {
+		// TODO(cnicolaou,toddw): consider using a new goroutine
+		// for every input record rather than fixing concurrency like
+		// this. Maybe an another option is to use the capacity of the
+		// input channel.
+		mr.NumMappers = runtime.NumCPU()
+	}
+	var timeout <-chan time.Time
+	if mr.Timeout > 0 {
+		timeout = time.After(mr.Timeout)
+	}
+	defer close(mr.output)
+	if mr.err = mr.runMappers(mapper, timeout); mr.err != nil {
+		return mr.err
+	}
+	mr.err = mr.runReducers(reducer, timeout)
+	return mr.err
+}
diff --git a/simplemr/mr_test.go b/simplemr/mr_test.go
new file mode 100644
index 0000000..00ed1f8
--- /dev/null
+++ b/simplemr/mr_test.go
@@ -0,0 +1,205 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package simplemr_test
+
+import (
+	"fmt"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+
+	"v.io/x/lib/simplemr"
+)
+
+func newChans(chanSize int) (chan *simplemr.Record, chan *simplemr.Record) {
+	return make(chan *simplemr.Record, chanSize), make(chan *simplemr.Record, chanSize)
+}
+
+type termCount struct{}
+
+func (tc *termCount) Map(mr *simplemr.MR, key string, val interface{}) error {
+	text, ok := val.(string)
+	if !ok {
+		return fmt.Errorf("%T is the wrong type", val)
+	}
+	for _, token := range strings.Split(text, " ") {
+		mr.MapOut(token, 1)
+	}
+	return nil
+}
+
+func (tc *termCount) Reduce(mr *simplemr.MR, key string, values []interface{}) error {
+	count := 0
+	for _, val := range values {
+		c, ok := val.(int)
+		if !ok {
+			return fmt.Errorf("%T is the wrong type", val)
+		}
+		count += c
+	}
+	mr.ReduceOut(key, count)
+	return nil
+}
+
+var (
+	d1 = "a b c"
+	d2 = "a b c d"
+	d3 = "e f"
+)
+
+func expect(t *testing.T, out chan *simplemr.Record, key string, vals ...int) {
+	rec := <-out
+	if got, want := rec.Key, key; got != want {
+		t.Errorf("got %v, want %v", got, want)
+	}
+	if got, want := len(rec.Values), len(vals); got != want {
+		t.Errorf("got %v, want %v", got, want)
+		return
+	}
+	for i, v := range vals {
+		if got, want := rec.Values[i], v; got != want {
+			t.Errorf("%d: got %v, want %v", i, got, want)
+		}
+	}
+}
+
+func TestMR(t *testing.T) {
+	mrt := &simplemr.MR{}
+	in, out := newChans(10)
+	tc := &termCount{}
+	genInput := func() {
+		in <- &simplemr.Record{"d1", []interface{}{d1, d2, d3}}
+		in <- &simplemr.Record{"d2", []interface{}{d1, d2, d3}}
+		close(in)
+	}
+	go genInput()
+	if err := mrt.Run(in, out, tc, tc); err != nil {
+		t.Fatal(err)
+	}
+
+	expect(t, out, "a", 4)
+	expect(t, out, "b", 4)
+	expect(t, out, "c", 4)
+	expect(t, out, "d", 2)
+	expect(t, out, "e", 2)
+	expect(t, out, "f", 2)
+	kvs := <-out
+	if kvs != nil {
+		t.Fatal("expected the channel to be closed")
+	}
+}
+
+type slowReducer struct{}
+
+func (sr *slowReducer) Reduce(mr *simplemr.MR, key string, values []interface{}) error {
+	time.Sleep(time.Hour)
+	return nil
+}
+
+func TestTimeout(t *testing.T) {
+	in, out := newChans(1)
+	mrt := &simplemr.MR{Timeout: 10 * time.Millisecond}
+	identity := &simplemr.Identity{}
+	mrt.Run(in, out, identity, identity)
+	if err := mrt.Error(); err == nil || !strings.Contains(err.Error(), "timed out mappers") {
+		t.Fatalf("missing or wrong error: %v", err)
+	}
+	mrt = &simplemr.MR{Timeout: 10 * time.Millisecond}
+	in, out = newChans(1)
+	in <- &simplemr.Record{"key", []interface{}{"value"}}
+	close(in)
+	mrt.Run(in, out, identity, &slowReducer{})
+	if err := mrt.Error(); err == nil || !strings.Contains(err.Error(), "timed out reducers") {
+		t.Fatalf("missing or wrong error: %v", err)
+	}
+}
+
+type sleeper struct{}
+
+const sleepTime = time.Millisecond * 100
+
+func (sl *sleeper) Map(mr *simplemr.MR, key string, val interface{}) error {
+	time.Sleep(sleepTime)
+	mr.MapOut(key, val)
+	return nil
+}
+
+func (sl *sleeper) Reduce(mr *simplemr.MR, key string, values []interface{}) error {
+	mr.ReduceOut(key, values...)
+	return nil
+}
+
+func runMappers(t *testing.T, bufsize, numMappers int) time.Duration {
+	mrt := &simplemr.MR{NumMappers: numMappers}
+	in, out := newChans(bufsize)
+	sl := &sleeper{}
+	go func() {
+		for i := 0; i < bufsize; i++ {
+			in <- &simplemr.Record{Key: fmt.Sprintf("%d", i), Values: []interface{}{i}}
+		}
+		close(in)
+	}()
+	then := time.Now()
+	if err := mrt.Run(in, out, sl, sl); err != nil {
+		t.Fatal(err)
+	}
+	return time.Since(then)
+}
+
+func TestOneMappers(t *testing.T) {
+	bufsize := 5
+	runtime := runMappers(t, bufsize, 1)
+	if got, want := runtime, time.Duration(int64(sleepTime)*int64(bufsize)); got < want {
+		t.Errorf("took %s which is too fast, should be at least %s", got, want)
+	}
+}
+
+func TestMultipleMappers(t *testing.T) {
+	numCPUs := runtime.NumCPU()
+	if numCPUs == 1 {
+		t.Skip("can't test concurrency with only one CPU")
+	}
+	bufsize := 5
+	runtime := runMappers(t, bufsize, numCPUs)
+	if got, want := runtime, time.Duration(int64(sleepTime)*int64(bufsize)); got > want {
+		t.Errorf("took %s which is too slow, should take no longer than %s", got, want)
+	}
+}
+
+type adder struct{}
+
+func (a *adder) Map(mr *simplemr.MR, key string, val interface{}) error {
+	i := val.(int)
+	i++
+	mr.MapOut(key, i)
+	return nil
+}
+
+func (a *adder) Reduce(mr *simplemr.MR, key string, values []interface{}) error {
+	mr.ReduceOut(key, values...)
+	return nil
+}
+
+func TestChainedMR(t *testing.T) {
+	chanSize := 5
+	in, middle, out := make(chan *simplemr.Record, chanSize), make(chan *simplemr.Record, chanSize), make(chan *simplemr.Record, chanSize)
+	mrt1 := &simplemr.MR{}
+	mrt2 := &simplemr.MR{}
+	adder := &adder{}
+	go mrt1.Run(in, middle, adder, adder)
+	go mrt2.Run(middle, out, adder, adder)
+	in <- &simplemr.Record{"1", []interface{}{1}}
+	in <- &simplemr.Record{"2", []interface{}{2}}
+	close(in)
+	expect(t, out, "1", 3)
+	expect(t, out, "2", 4)
+	if err := mrt1.Error(); err != nil {
+		t.Fatal(err)
+	}
+	if err := mrt2.Error(); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/simplemr/util.go b/simplemr/util.go
new file mode 100644
index 0000000..8d1a19a
--- /dev/null
+++ b/simplemr/util.go
@@ -0,0 +1,17 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package simplemr
+
+type Identity struct{}
+
+func (i *Identity) Map(mr *MR, key string, val interface{}) error {
+	mr.MapOut(key, val)
+	return nil
+}
+
+func (i *Identity) Reduce(mr *MR, key string, values []interface{}) error {
+	mr.ReduceOut(key, values...)
+	return nil
+}