fix all bugs : package mr

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

type TaskStatus int

const (
 NotStarted TaskStatus = iota
 InProgress
 Completed
)

type Coordinator struct {
 NReduce int // Number of reduce tasks to be performed
 InputFiles []string // A list of input files to be processed
 MapTaskSt map[string]TaskStatus // A map of map task numbers to their current status
 ReduceTaskSt map[int]TaskStatus // A map of reduce task numbers to their current status
 Workers map[string]WorkerStatus // A map of worker IDs to their current status
 IntermediateFiles [][]string // 2D Slice that containst the intermediate file names produced by map tasks, each row represents a different reduce task
 MapIndex int // Next map task to be assigned
 ReduceIndex int // Next reduce task to be assigned
 MapFinished bool // True when all map tasks are completed
 ReduceFinished bool // True when all reduce tasks are completed
 Finish bool // True when all map and reduce tasks are completed
 WriterLock \*sync.RWMutex // Lock for controlling shared variables
 MapTasks chan MapTask // Channel for uncompleted map tasks
 ReduceTasks chan int // Channel for uncompleted reduce tasks
}

// Starting Coordinator Logic
func (c \*Coordinator) Start() {
 fmt.Println("Starting Coordinator, adding Map Tasks to channel")
 //Prepare initial MapTasks and add them to the queue
 for index, file := range c.InputFiles {
 c.MapTaskSt[file] = NotStarted
 mapTask := MapTask{
 FileName: file,
 NReduce: c.NReduce,
 Index: index,
 }
 c.MapTasks <- mapTask
 fmt.Println("MapTask", mapTask, "added to channel")
 }
 c.server()
}

func (c \*Coordinator) getNextMapTask() (string, TaskStatus) {
 for file, status := range c.MapTaskSt {
 if status == NotStarted {
 return file, status
 }
 }
 return "", Completed
}

// Your code here -- RPC handlers for the worker to call.
// an example RPC handler.
// the RPC argument and reply types are defined in rpc.go.
func (c \*Coordinator) AssignMapTask(args \*EmptyArgs, reply \*MapTaskReply) error {
 fmt.Println("Map Task Requested")
 c.WriterLock.Lock()
 defer c.WriterLock.Unlock()

 if !c.MapFinished {
 // Find an Uncompleted Map Task
 file, status := c.getNextMapTask()
 if status == NotStarted {
 task := <-c.MapTasks
 fmt.Println("Map task found", file)
 c.MapTaskSt[file] = InProgress
 reply.Filename = file
 reply.NReduce = c.NReduce
 reply.Index = c.MapIndex
 c.MapIndex++
 go c.WaitForMapWorker(task)
 return nil
 }
 }
 // No more map Tasks to Assign
 reply.Filename = ""
 reply.NReduce = c.NReduce
 reply.Index = -1
 return nil
}

func (c \*Coordinator) getNextReduceTask() (int, TaskStatus) {
 for int, status := range c.ReduceTaskSt {
 if status == NotStarted {
 return int, status
 }
 }
 return -1, Completed
}
func (c \*Coordinator) AssignReduceTask(args \*EmptyArgs, reply \*ReduceTaskReply) error {
 fmt.Println("Reduce Task Requested")
 c.WriterLock.Lock()
 defer c.WriterLock.Unlock()

 if !c.ReduceFinished {
 // Find the next uncompleted Reduce Task
 i, status := c.getNextReduceTask()
 if status == NotStarted {
 NumReduce := <-c.ReduceTasks
 reply.IntermediateFiles = c.IntermediateFiles[NumReduce]
 fmt.Println("NumReduce:", NumReduce, "Files:", c.IntermediateFiles[NumReduce])
 reply.Index = i
 c.ReduceTaskSt[NumReduce] = InProgress
 go c.WaitForReduceWorker(NumReduce)
 return nil
 }
 }
 // No more reduce tasks to assign
 fmt.Println("No more reduce")
 reply.IntermediateFiles = []string{}
 reply.Index = -1
 return nil
}

// Goroutine will wait 10 seconds and check if map task is completed or not
func (c \*Coordinator) WaitForMapWorker(task MapTask) {
 time.Sleep(time.Second \* 10)
 c.WriterLock.Lock()
 if c.MapTaskSt[task.FileName] == NotStarted {
 fmt.Println("Timer expired, task", task.FileName, "is not finished. Putting back in queue")
 c.MapTasks <- task
 } else {
 if c.MapTaskSt[task.FileName] == Completed {
 c.WriterLock.Unlock()
 return
 }
 }
 c.WriterLock.Unlock()
}

// Goroutine will wait 10 seconds and check if reduce task is completed or not
func (c \*Coordinator) WaitForReduceWorker(NumReduce int) {
 time.Sleep(time.Second \* 10)
 c.WriterLock.Lock()
 if c.ReduceTaskSt[NumReduce] == NotStarted {
 fmt.Println("Timer expired, reduce task number", NumReduce, "is not finished. Putting back in queue")
 c.ReduceTasks <- NumReduce
 } else {
 if c.ReduceTaskSt[NumReduce] == Completed {
 c.WriterLock.Unlock()
 return
 }
 }
 c.WriterLock.Unlock()
}

func (c \*Coordinator) MapTaskCompleted(args \*MapTaskReply, reply \*EmptyReply) error {
 c.WriterLock.Lock()
 defer c.WriterLock.Unlock()

 c.MapTaskSt[args.Filename] = Completed

 for \_, status := range c.MapTaskSt {
 if status != Completed {
 return nil
 }
 }

 c.MapFinished = true
 if c.MapFinished {
 for x := 0; x < c.NReduce; x++ {
 c.ReduceTasks <- x
 c.ReduceTaskSt[x] = NotStarted
 }
 }

 // fmt.Println("Task", args, "completed")
 return nil
}

func (c \*Coordinator) ReduceTaskCompleted(args \*ReduceTaskCompleted, reply \*EmptyReply) error {
 c.WriterLock.Lock()
 defer c.WriterLock.Unlock()

 if \_, ok := c.ReduceTaskSt[args.Index]; ok {
 c.ReduceTaskSt[args.Index] = Completed

 Finished := true
 for \_, status := range c.ReduceTaskSt {
 if status != Completed {
 return nil
 }
 }
 if Finished {
 c.ReduceFinished = true
 }
 } else {
 log.Printf("Invalid reduce task index: %d", args.Index)
 }
 return nil
}

func (c \*Coordinator) IntermediateFilesCompleted(args \*IntermediateFilesArgs, reply \*EmptyReply) error {
 c.WriterLock.Lock()
 defer c.WriterLock.Unlock()

 c.IntermediateFiles[args.Index] = append(c.IntermediateFiles[args.Index], args.Filename)
 return nil
}

// 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 {
 // Acquire writer lock to protect shared data
 c.WriterLock.Lock()
 defer c.WriterLock.Unlock()

 // Have all Map and Reduce Tasks been completed
 c.Finish = c.MapFinished && c.ReduceFinished
 return c.Finish

}

// 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{
 MapTaskSt: make(map[string]TaskStatus),
 ReduceTaskSt: make(map[int]TaskStatus),
 MapFinished: false,
 ReduceFinished: false,
 Finish: false,
 WriterLock: &sync.RWMutex{},
 Workers: make(map[string]WorkerStatus),
 NReduce: NReduce,
 InputFiles: files,
 IntermediateFiles: make([][]string, NReduce),
 MapIndex: 0,
 ReduceIndex: 0,
 MapTasks: make(chan MapTask, len(files)),
 ReduceTasks: make(chan int, NReduce),
 }

 fmt.Println("Calling Start()")
 c.Start() // Calls Start Function

 return &c
}
모든 버그 수정 : 패키지 미스터임포트 ( "fmt" "log" "net" "net/http" "net/rpc" "os" "sync" "time")유형 TaskStatus intconst ( NotStarted TaskStatus = iota InProgress Completed)유형 코디네이터 구조체 { NReduce int // 수행해야 할 감소 작업 수 InputFiles []string // 처리할 입력 파일 목록 MapTaskSt map[string]TaskStatus // 태스크 번호를 현재 상태에 매핑한 맵입니다. ReduceTaskSt map[int]TaskStatus // 작업 번호를 현재 상태로 줄인 맵입니다. Workers map[string]WorkerStatus // 작업자 ID를 현재 상태로 매핑한 맵입니다. IntermediateFiles [][]string // 맵 작업에서 생성된 중간 파일 이름을 포함하는 2D 슬라이스, 각 행은 다른 리듀스 작업을 나타냅니다. MapIndex int // 다음에 할당할 맵 작업입니다. ReduceIndex int // 다음에 할당할 리듀스 태스크 MapFinished 부울 // 모든 맵 작업이 완료되면 참입니다. ReduceFinished bool // 모든 감소 작업이 완료되면 참입니다. Finish bool // 모든 맵 작업과 리듀스 작업이 완료되면 참입니다. WriterLock \*sync.RWMutex // 공유 변수 제어를 위한 잠금 MapTasks chan MapTask // 완료되지 않은 맵 작업을 위한 채널 ReduceTasks chan int // 완료되지 않은 감소 태스크를 위한 채널}// 코디네이터 로직 시작func (c \*Coordinator) Start() { fmt.Println("코디네이터 시작, 채널에 맵 태스크 추가") //P 초기 MapTask를 준비하고 큐에 추가합니다. for index, file := 범위 c.InputFiles { c.MapTaskSt[file] = NotStarted mapTask := MapTask{ 파일 이름: 파일, NReduce: c.NReduce, Index: index, } c.MapTasks <- mapTask fmt.Println("MapTask", mapTask, "채널에 추가됨") } c.server()}func (c \*Coordinator) getNextMapTask() (string, TaskStatus) { for file, status := range c.MapTaskSt { if status == NotStarted { 반환 파일, 상태 } } 반환 "", 완료}// 여기 코드 -- 워커가 호출할 RPC 핸들러입니다.// 예제 RPC 핸들러입니다.// RPC 인자와 응답 유형은 rpc.go에 정의되어 있습니다.func (c \*Coordinator) AssignMapTask(args \*EmptyArgs, reply \*MapTaskReply) error { fmt.Println("지도 작업 요청됨") c.WriterLock.Lock() defer c.WriterLock.Unlock() if !c.MapFinished { // 완료되지 않은 지도 태스크 찾기 파일, status := c.getNextMapTask() if status == NotStarted { task := <-c.MapTasks fmt.Println("지도 작업 발견", file) c.MapTaskSt[file] = InProgress reply.Filename = 파일 reply.NReduce = c.NReduce reply.Index = c.MapIndex c.MapIndex++ go c.WaitForMapWorker(task) 반환 nil } } // 더 이상 할당할 지도 작업 없음 reply.Filename = "" reply.NReduce = c.NReduce reply.Index = -1 반환 nil}func (c \*Coordinator) getNextReduceTask() (int, TaskStatus) { for int, status := range c.ReduceTaskSt { if status == NotStarted { 반환 int, status } } 반환 -1, 완료}func (c \*Coordinator) AssignReduceTask(args \*EmptyArgs, reply \*ReduceTaskReply) error { fmt.Println("요청된 작업 감소") c.WriterLock.Lock() defer c.WriterLock.Unlock() if !c.ReduceFinished { // 다음 완료되지 않은 감소 태스크 찾기 i, status := c.getNextReduceTask() if status == NotStarted { NumReduce := <-c.ReduceTasks reply.IntermediateFiles = c.IntermediateFiles[NumReduce] fmt.Println("NumReduce:", NumReduce, "Files:", c.IntermediateFiles[NumReduce]) reply.Index = i c.ReduceTaskSt[NumReduce] = InProgress go c.WaitForReduceWorker(NumReduce) 반환 nil } } // 더 이상 할당할 작업 감소가 없습니다. fmt.Println("더 이상 줄일 작업이 없습니다") reply.IntermediateFiles = []string{} reply.Index = -1 반환 nil}// 고루틴은 10초를 기다렸다가 지도 작업이 완료되었는지 확인합니다.func (c \*Coordinator) WaitForMapWorker(task MapTask) { time.Sleep(time.Second \* 10) c.WriterLock.Lock() c.MapTaskSt[task.FileName] == NotStarted { fmt.Println("타이머가 만료되었습니다, 작업", task.FileName, "완료되지 않았습니다. 대기열에 다시 넣습니다.") c.MapTasks <- task } else { c.MapTaskSt[task.FileName] == Completed { c.WriterLock.Unlock() return } } c.WriterLock.Unlock()}// 고루틴은 10초를 기다렸다가 작업 감소가 완료되었는지 여부를 확인합니다.func (c \*Coordinator) WaitForReduceWorker(NumReduce int) { time.Sleep(time.Second \* 10) c.WriterLock.Lock() c.ReduceTaskSt[NumReduce] == NotStarted { fmt.Println("타이머가 만료되었습니다, 작업 수를 줄이십시오", NumReduce, "완료되지 않았습니다. 다시 대기열에 넣습니다.") c.ReduceTasks <- NumReduce } else { if c.ReduceTaskSt[NumReduce] == Completed { c.WriterLock.Unlock() return } } c.WriterLock.Unlock()}func (c \*Coordinator) MapTaskCompleted(args \*MapTaskReply, reply \*EmptyReply) error { c.WriterLock.Lock() defer c.WriterLock.Unlock() c.MapTaskSt[args.Filename] = Completed for \_, status := 범위 c.MapTaskSt { if status != Completed { 반환 nil } } c.MapFinished = true if c.MapFinished { for x := 0; x < c.NReduce; x++ { c.ReduceTasks <- x c.ReduceTaskSt[x] = NotStarted } } // fmt.Println("Task", args, "completed") 반환 nil}func (c \*Coordinator) ReduceTaskCompleted(args \*ReduceTaskCompleted, reply \*Em)