شمارش کلمات با سرعت رعد و برق: کانال های Golang و استخرهای کارگر برای پردازش پرونده های متنی

package main

import (
    "fmt"
    "strings"
    "time"
)

func countWords(filename string, content string) int {
    time.Sleep(100 * time.Millisecond) // Simulate I/O
    return len(strings.Fields(content))
}

func main() {
    // Each file has exactly 11 words
    const testContent = "this is a sample text file that has eleven words here"

    files := make(map[string]string)
    for i := 1; i <= 20; i++ {
        files[fmt.Sprintf("file%d.txt", i)] = testContent
    }

    start := time.Now()
    total := 0
    for filename, content := range files {
        count := countWords(filename, content)
        total += count
    }
    fmt.Printf("Total words: %d, Time taken: %v\n", total, time.Since(start))
}

Total words: 220, Time taken: 2.01s

package main

import (
    "fmt"
    "strings"
    "time"
)

func countWords(filename string, content string) int {
    time.Sleep(100 * time.Millisecond)
    return len(strings.Fields(content))
}

type Job struct {
    filename string
    content  string
}

func main() {
    // Each file has exactly 11 words
    const testContent = "this is a sample text file that has eleven words here"

    files := make(map[string]string)
    for i := 1; i <= 20; i++ {
        files[fmt.Sprintf("file%d.txt", i)] = testContent
    }

    jobs := make(chan Job)
    done := make(chan bool)

    go func() {
        total := 0
        for job := range jobs {
            count := countWords(job.filename, job.content)
            total += count
        }
        fmt.Printf("Worker total: %d\n", total)
        done <- true
    }()

    start := time.Now()
    for filename, content := range files {
        jobs <- Job{filename, content}
    }
    close(jobs)

    <-done // Wait for worker to finish
    fmt.Printf("Time taken: %v\n", time.Since(start))
}

Worker total: 220
Time taken: 2.01s

package main

import (
    "fmt"
    "strings"
    "sync"
    "time"
)

func worker(id int, jobs <-chan Job, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        count := len(strings.Fields(job.content))
        time.Sleep(100 * time.Millisecond)
        results <- count
    }
}

type Job struct {
    filename string
    content  string
}

func main() {
    // Each file has exactly 11 words
    const testContent = "this is a sample text file that has eleven words here"

    files := make(map[string]string)
    for i := 1; i <= 20; i++ {
        files[fmt.Sprintf("file%d.txt", i)] = testContent
    }

    numWorkers := 3
    numJobs := len(files)

    jobs := make(chan Job, numJobs)       // Buffered jobs channel
    results := make(chan int, numJobs)    // Buffered results channel
    var wg sync.WaitGroup

    // Start workers
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // Send all jobs
    start := time.Now()
    for filename, content := range files {
        jobs <- Job{filename, content}
    }
    close(jobs)  // Close jobs after sending all work

    // Wait for all workers to finish in a separate goroutine
    go func() {
        wg.Wait()
        close(results)  // Close results only after all workers are done
    }()

    // Collect results
    total := 0
    for count := range results {  // This will exit when results channel is closed
        total += count
    }

    fmt.Printf("Total words: %d, Time taken: %v\n", total, time.Since(start))
}

Total words: 220, Time taken: 704ms

package main

import (
    "fmt"
    "strings"
    "sync"
    "time"
)

func worker(id int, jobs <-chan Job, results chan<- int) {
    for job := range jobs {
        count := len(strings.Fields(job.content))
        time.Sleep(100 * time.Millisecond)
        results <- count
    }
}

type Job struct {
    filename string
    content  string
}

func main() {
    // Each file has exactly 11 words
    const testContent = "this is a sample text file that has eleven words here"

    files := make(map[string]string)
    for i := 1; i <= 20; i++ {
        files[fmt.Sprintf("file%d.txt", i)] = testContent
    }

    jobs := make(chan Job, 20)     // Buffered channel
    results := make(chan int, 20)  // Buffered results channel
    var wg sync.WaitGroup

    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func(workerId int) {
            defer wg.Done()
            worker(workerId, jobs, results)
        }(i)
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    start := time.Now()
    for filename, content := range files {
        jobs <- Job{filename, content}
    }
    close(jobs)

    total := 0
    for count := range results {
        total += count
    }
    fmt.Printf("Total words: %d, Time taken: %v\n", total, time.Since(start))
}

Total words: 220, Time taken: 703ms

package main

import (
    "fmt"
    "strings"
    "sync"
    "time"
)

func worker(id int, jobs <-chan Job, results chan<- int) {
    for job := range jobs {
        count := len(strings.Fields(job.content))
        time.Sleep(100 * time.Millisecond)
        results <- count
    }
}

type Job struct {
    filename string
    content  string
}

func processFiles(files map[string]string, numWorkers int) (int, time.Duration) {
    numFiles := len(files)
    jobs := make(chan Job, numFiles)     // Buffer all files
    results := make(chan int, numFiles)   // Buffer all results
    var wg sync.WaitGroup

    // Start workers
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go func(workerId int) {
            defer wg.Done()
            worker(workerId, jobs, results)
        }(i)
    }

    // Close results when all workers are done
    go func() {
        wg.Wait()
        close(results)
    }()

    // Send all jobs
    start := time.Now()
    for filename, content := range files {
        jobs <- Job{filename, content}
    }
    close(jobs)

    // Collect results
    total := 0
    for count := range results {
        total += count
    }
    return total, time.Since(start)
}

func main() {
    // Generate test files
    files := make(map[string]string)
    for i := 1; i <= 20; i++ {
        files[fmt.Sprintf("file%d.txt", i)] = "this is a sample text file with some words to count"
    }

    // Try different numbers of workers
    for _, workers := range []int{1, 2, 3, 4, 5} {
        total, duration := processFiles(files, workers)
        fmt.Printf("%d workers - Total: %d, Time: %v\n", 
            workers, total, duration)
    }
}

1 worker  - Total: 180, Time: 2.01s
2 workers - Total: 180, Time: 1.02s
3 workers - Total: 180, Time: 704ms
4 workers - Total: 180, Time: 602ms
5 workers - Total: 180, Time: 503ms

package main

import (
    "fmt"
    "math/rand"
    "strings"
    "sync"
    "time"
)

type Job struct {
    filename string
    content  string
}

func generateFiles(n int) map[string]string {
    files := make(map[string]string)
    words := []string{"the", "quick", "brown", "fox", "jumps", "over", "lazy", "dog",
        "pack", "my", "box", "with", "five", "dozen", "liquor", "jugs"}

    r := rand.New(rand.NewSource(time.Now().UnixNano()))

    for i := 1; i <= n; i++ {
        filename := fmt.Sprintf("file%d.txt", i)
        var content []string
        // Generate files with varying sizes (10-50 words)
        wordCount := r.Intn(41) + 10
        for j := 0; j < wordCount; j++ {
            content = append(content, words[r.Intn(len(words))])
        }
        files[filename] = strings.Join(content, " ")
    }
    return files
}

func processFiles(files map[string]string, numWorkers int) (int, time.Duration) {
    numFiles := len(files)
    jobs := make(chan Job, numFiles)
    results := make(chan int, numFiles)
    var wg sync.WaitGroup

    start := time.Now()

    // Start workers
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go func(workerId int) {
            defer wg.Done()
            for job := range jobs {
                time.Sleep(100 * time.Millisecond) // Simulate I/O
                count := len(strings.Fields(job.content))
                results <- count
            }
        }(i)
    }

    // Send all jobs
    for filename, content := range files {
        jobs <- Job{filename, content}
    }
    close(jobs)

    // Wait for workers and close results
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    total := 0
    for count := range results {
        total += count
    }

    return total, time.Since(start)
}

func main() {
    // Generate 200 test files
    files := generateFiles(200)

    // Sequential processing
    start := time.Now()
    seqTotal := 0
    for _, content := range files {
        time.Sleep(100 * time.Millisecond) // Simulate I/O
        seqTotal += len(strings.Fields(content))
    }
    seqTime := time.Since(start)

    // Try different worker pool sizes
    workerCounts := []int{1, 5, 10, 20, 50}
    fmt.Printf("Sequential: %d words in %v\n", seqTotal, seqTime)

    for _, workers := range workerCounts {
        total, duration := processFiles(files, workers)
        speedup := float64(seqTime) / float64(duration)
        fmt.Printf("%2d workers: %d words in %v (%.2fx faster)\n", 
            workers, total, duration, speedup)
    }
}

Sequential: 6148 words in 20s
 1 workers: 6148 words in 20s (1.00x faster)
 5 workers: 6148 words in 4s (5.00x faster)
10 workers: 6148 words in 2s (10.00x faster)
20 workers: 6148 words in 1s (20.00x faster)
50 workers: 6148 words in 400ms (50.00x faster)