MIT6.824 Lab1

package mr

import (
	"bytes"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"path/filepath"
	"sync"
	"time"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	info := doGetInfo()
	log.Printf("Worker: receive coordinator's info %v\n", info)
	for {
		task := doCorrespond()
		log.Printf("Worker: receive coordinator's response %v\n", task)
		switch task.Type {
		case MapType:
			MapTask(mapf, task.File, info.NReduce, task.Id)
		case ReduceType:
			ReduceTask(reducef, info.NMap, task.Id)
		case WaitType:
			time.Sleep(1 * time.Second)
		case StopType:
			return
		default:
		}
	}
}

func MapTask(mapF func(string, string) []KeyValue, filePath string, NReduce int, id int) {
	fileName := filePath
	file, err := os.Open(fileName)
	if err != nil {
		log.Fatalf("cannot open %v", fileName)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", fileName)
	}
	file.Close()

	kva := mapF(fileName, string(content))
	intermediates := make([][]KeyValue, NReduce)
	for _, kv := range kva {
		index := ihash(kv.Key) % NReduce
		intermediates[index] = append(intermediates[index], kv)
	}

	var wg sync.WaitGroup
	for index, intermediate := range intermediates {
		wg.Add(1)
		go func(index int, intermediate []KeyValue) {
			defer wg.Done()
			tmpFileName := fmt.Sprintf("mr-%d-%d", id, index)
			var buf bytes.Buffer
			enc := json.NewEncoder(&buf)
			for _, kv := range intermediate {
				err := enc.Encode(&kv)
				if err != nil {
					log.Fatalf("cannot encode json %v\n", kv.Key)
				}
			}
			err := WriteFile(tmpFileName, &buf)
			if err != nil {
				log.Fatalf("cannot write file %v, %v\n", tmpFileName, err)
			}
		}(index, intermediate)
	}
	wg.Wait()
	doReport(id, MapType)
}

func ReduceTask(reduceF func(string, []string) string, NMap int, id int) {
	var kva []KeyValue
	for i := 0; i < NMap; i++ {
		filePath := fmt.Sprintf("mr-%d-%d", i, id)
		file, err := os.Open(filePath)
		if err != nil {
			log.Fatalf("cannot open %v", filePath)
		}
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kva = append(kva, kv)
		}
		file.Close()
	}
	results := make(map[string][]string)

	for _, kv := range kva {
		results[kv.Key] = append(results[kv.Key], kv.Value)
	}
	var buf bytes.Buffer
	for key, values := range results {
		output := reduceF(key, values)
		fmt.Fprintf(&buf, "%v %v\n", key, output)
	}

	tmpFileName := fmt.Sprintf("mr-out-%d", id)
	err := WriteFile(tmpFileName, &buf)
	if err != nil {
		log.Fatalf("cannot write file %v, %v\n", tmpFileName, err)
	}
	doReport(id, ReduceType)
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}

func doCorrespond() *CorrespondResponse {
	rsp := CorrespondResponse{}
	call("Coordinator.Correspond", &CorrespondRequest{}, &rsp)
	return &rsp
}

func doGetInfo() *InfoResponse {
	rsp := InfoResponse{}
	call("Coordinator.Info", &InfoRequest{}, &rsp)
	return &rsp
}

func doReport(id int, t TaskType) {
	call("Coordinator.Report", &ReportRequest{id, t}, &ReportResponse{})
}

func WriteFile(filename string, r io.Reader) error {
	dir, file := filepath.Split(filename)
	if dir == "" {
		dir = "."
	}

	f, err := ioutil.TempFile(dir, file)
	if err != nil {
		return fmt.Errorf("cannot create temp file: %v", err)
	}
	defer func() {
		if err != nil {
			_ = os.Remove(f.Name())
		}
	}()
	defer f.Close()
	name := f.Name()
	if _, err := io.Copy(f, r); err != nil {
		return fmt.Errorf("cannot write data to tempfile %q: %v", name, err)
	}
	if err := f.Close(); err != nil {
		return fmt.Errorf("can't close tempfile %q: %v", name, err)
	}

	info, err := os.Stat(filename)
	if os.IsNotExist(err) {
	} else if err != nil {
		return err
	} else {
		if err := os.Chmod(name, info.Mode()); err != nil {
			return fmt.Errorf("can't set filemode on tempfile %q: %v", name, err)
		}
	}
	if err := os.Rename(name, filename); err != nil {
		return fmt.Errorf("cannot replace %q with tempfile %q: %v", filename, name, err)
	}
	return nil
}

package mr

import (
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"sync"
	"time"
)

const (
	Timeout = time.Second * 10
)

var once *sync.Once

type Task struct {
	File      string
	Id        int
	Type      TaskType
	Status    TaskStatus
	StartTime time.Time
}

type Coordinator struct {
	lock sync.Mutex

	Files []string
	// Your definitions here.
	NReduce  int
	NMap     int
	Close    bool
	TaskCh   chan CorrespondMsg
	Tasks    []Task
	doneCh   chan struct{}
	ReportCh chan ReportMsg
	WorkFlow Flow
}

type CorrespondMsg struct {
	response *CorrespondResponse
	ok       chan struct{}
}

type ReportMsg struct {
	request *ReportRequest
	ok      chan struct{}
}

// Your code here -- RPC handlers for the worker to call.

func (c *Coordinator) Correspond(request *CorrespondRequest, response *CorrespondResponse) error {
	msg := CorrespondMsg{response, make(chan struct{})}
	c.TaskCh <- msg
	<-msg.ok // wait send task response
	return nil
}

func (c *Coordinator) Info(request *InfoRequest, response *InfoResponse) error {
	response.NMap, response.NReduce, response.Close = c.NMap, c.NReduce, c.Close
	return nil
}

func (c *Coordinator) Report(request *ReportRequest, response *ReportResponse) error {
	msg := ReportMsg{request, make(chan struct{})}
	c.ReportCh <- msg
	<-msg.ok
	return nil
}

func (c *Coordinator) Scheduler() {
	for {
		switch c.WorkFlow {
		case OnInit:
			c.WorkFlow = OnMap
		case OnMap:
			once.Do(func() {
				log.Printf("Coordinator start %v\n", OnMap)
				c.initMap()
			})
			select {
			case msg := <-c.TaskCh:
				c.Assigner(&msg)
				log.Printf("Coordinator: assigned a task %v to worker\n", msg.response)
				msg.ok <- struct{}{}
			case msg := <-c.ReportCh:
				log.Printf("Coordinator: Worker has executed task %v \n", msg.request)
				c.lock.Lock()
				c.Tasks[msg.request.Id].Status = Finished
				c.lock.Unlock()
				msg.ok <- struct{}{}
			}
		case OnReduce:
			once.Do(func() {
				log.Printf("Coordinator: %v finished, start %v\n", OnMap, OnReduce)
				c.initReduce()
			})
			select {
			case msg := <-c.TaskCh:
				c.Assigner(&msg)
				log.Printf("Coordinator: assigned a task %v to worker\n", msg.response)
				msg.ok <- struct{}{}
			case msg := <-c.ReportCh:
				log.Printf("Coordinator: Worker has executed task %v\n", msg.request)
				c.lock.Lock()
				c.Tasks[msg.request.Id].Status = Finished
				c.lock.Unlock()
				msg.ok <- struct{}{}
			}
		case OnComplete:
			once.Do(func() {
				log.Printf("Coordinator: %v finished\n", OnReduce)
				c.initComplete()
			})
			select {
			case msg := <-c.TaskCh:
				msg.response.Type = StopType
				log.Printf("Coordinator: assigned stop to worker\n")
				msg.ok <- struct{}{}
			}
		}
	}
}

func (c *Coordinator) Assigner(msg *CorrespondMsg) {
	c.lock.Lock()
	defer c.lock.Unlock()

	finish := true
	getJob := false
	for id, task := range c.Tasks {
		switch task.Status {
		case UnDo:
			finish = false
			getJob = true
			c.Tasks[id].Status = Working
			c.Tasks[id].StartTime = time.Now()
			msg.response.Id = id
			switch c.WorkFlow {
			case OnMap:
				msg.response.Type = MapType
				msg.response.File = c.Files[id]
			case OnReduce:
				msg.response.Type = ReduceType
			default:
			}
		case Working:
			finish = false
			// set timeout
			if time.Now().Sub(task.StartTime) > Timeout {
				c.Tasks[id].Status = UnDo
			}
		case Finished:
		default:
		}
		if getJob {
			break
		}
	}
	if !getJob {
		msg.response.Type = WaitType
	}
	if finish {
		c.WorkFlow++
		once = new(sync.Once)
	}
}

func (c *Coordinator) initMap() {
	c.WorkFlow = OnMap
	c.Tasks = make([]Task, len(c.Files))
	for index, file := range c.Files {
		c.Tasks[index] = Task{
			File:   file,
			Id:     index,
			Status: UnDo,
			Type:   MapType,
		}
	}
}

func (c *Coordinator) initReduce() {
	c.WorkFlow = OnReduce
	c.Tasks = make([]Task, c.NReduce)
	for i := 0; i < c.NReduce; i++ {
		c.Tasks[i] = Task{
			Id:     i,
			Status: UnDo,
			Type:   ReduceType,
		}
	}
}

func (c *Coordinator) initComplete() {
	c.WorkFlow = OnComplete
	c.doneCh <- struct{}{}
}

//
// start a thread that listens for RPCs from worker.go
//
func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := coordinatorSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
//
func (c *Coordinator) Done() bool {
	<-c.doneCh
	c.Close = true
	return true
}

//
// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{
		Files:    files,
		NReduce:  nReduce,
		NMap:     len(files),
		Close:    false,
		doneCh:   make(chan struct{}, 1),
		TaskCh:   make(chan CorrespondMsg),
		ReportCh: make(chan ReportMsg),
		WorkFlow: OnInit,
	}
	// Your code here.
	once = new(sync.Once)
	c.server()
	go c.Scheduler()
	return &c
}

package mr

//
// RPC definitions.
//
// remember to capitalize all names.
//

import (
	"fmt"
	"os"
)
import "strconv"

type TaskType int

type Flow int

type TaskStatus int

const (
	WaitType TaskType = iota
	MapType
	ReduceType
	StopType
)

func (task TaskType) String() string {
	switch task {
	case MapType:
		return "MapType"
	case ReduceType:
		return "ReduceType"
	case WaitType:
		return "WaitType"
	case StopType:
		return "StopType"
	}
	panic(fmt.Sprintf("unexpected TaskType %d", task))
}

const (
	// TaskStatus
	UnDo TaskStatus = iota
	Working
	Finished
)

func (status TaskStatus) String() string {
	switch status {
	case UnDo:
		return "UnDo"
	case Working:
		return "Working"
	case Finished:
		return "Finished"
	}
	panic(fmt.Sprintf("unexpected TaskStatus %d", status))
}

const (
	// WorkFlow
	OnInit Flow = iota
	OnMap
	OnReduce
	OnComplete
)

func (flow Flow) String() string {
	switch flow {
	case OnInit:
		return "OnInit"
	case OnMap:
		return "OnMap"
	case OnReduce:
		return "OnReduce"
	case OnComplete:
		return "OnComplete"
	}
	panic(fmt.Sprintf("unexpected WorkFlow %d", flow))
}

type CorrespondRequest struct {
}

type CorrespondResponse struct {
	File string
	Id   int
	Type TaskType
}

type InfoRequest struct {
}

type InfoResponse struct {
	NMap    int
	NReduce int
	Close   bool
}

type ReportRequest struct {
	Id   int
	Type TaskType
}

type ReportResponse struct {
}

func (response CorrespondResponse) String() string {
	switch response.Type {
	case MapType:
		return fmt.Sprintf("{Type:%v,File:%v,Id:%v}", response.Type, response.File, response.Id)
	case ReduceType:
		return fmt.Sprintf("{Type:%v,Id:%v}", response.Type, response.Id)
	case WaitType, StopType:
		return fmt.Sprintf("{Type:%v}", response.Type)
	}
	panic(fmt.Sprintf("unexpected Type %d", response.Type))
}

func (request ReportRequest) String() string {
	return fmt.Sprintf("{Id:%v,Type:%v}", request.Id, request.Type)
}

// Add your RPC definitions here.

// Cook up a unique-ish UNIX-domain socket name
// in /var/tmp, for the coordinator.
// Can't use the current directory since
// Athena AFS doesn't support UNIX-domain sockets.
func coordinatorSock() string {
	s := "/var/tmp/824-mr-"
	s += strconv.Itoa(os.Getuid())
	return s
}