Concurrency

The Concurrency module provides comprehensive concurrency and parallelism capabilities for Soul, including goroutines, channels, mutexes, wait groups, promises, and other synchronization primitives for building concurrent applications.

Basic Concurrency

spawn - Execute function in goroutine

Spawn a function to run in a separate goroutine and return a promise:

// Simple spawned function
promise = Concurrency.spawn(soul() {
    println("Running in goroutine")
    return "completed"
})

// Wait for result
result = await promise
println(result)  // "completed"

run - Fire-and-forget execution

Run a function in a goroutine without waiting for results:

// Fire-and-forget execution
Concurrency.run(soul() {
    println("Background task running")
    // Do some work
})

// Continue with other work immediately
println("Main thread continues")

spawn with arguments

Pass arguments to spawned functions:

soul processData(data, multiplier) {
    return data * multiplier
}

promise = Concurrency.spawn(processData, 10, 2)
result = await promise
println(result)  // 20

Channels

createChannel - Create channels for communication

Create channels for goroutine communication:

// Unbuffered channel
ch = Concurrency.createChannel()

// Buffered channel
bufferedCh = Concurrency.createChannel(10)

// Send and receive
Concurrency.spawn(soul() {
    ch.send("hello")
})

message = ch.receive()
println(message)  // "hello"

Channel operations

Use channels for producer-consumer patterns:

soul producer(ch) {
    for (i = 0; i < 5; i++) {
        ch.send("item " + i)
        println("Sent: item " + i)
    }
    ch.close()
}

soul consumer(ch) {
    while (true) {
        try {
            item = ch.receive()
            println("Received: " + item)
        } catch (error) {
            if (error.message.contains("closed")) {
                break
            }
        }
    }
}

channel = Concurrency.createChannel(3)
Concurrency.spawn(producer, channel)
Concurrency.spawn(consumer, channel)

Synchronization Primitives

Mutex - Mutual exclusion

Use mutexes for protecting shared resources:

mutex = Concurrency.createMutex()
sharedCounter = 0

soul incrementCounter() {
    mutex.lock()
    try {
        sharedCounter = sharedCounter + 1
        println("Counter: " + sharedCounter)
    } finally {
        mutex.unlock()
    }
}

// Spawn multiple goroutines
for (i = 0; i < 5; i++) {
    Concurrency.spawn(incrementCounter)
}

tryLock - Non-blocking lock

Try to acquire a lock without blocking:

mutex = Concurrency.createMutex()

soul tryWork() {
    if (mutex.tryLock()) {
        try {
            println("Got lock, doing work")
            // Do work
        } finally {
            mutex.unlock()
        }
    } else {
        println("Could not get lock, skipping work")
    }
}

Concurrency.spawn(tryWork)

WaitGroup - Wait for multiple goroutines

Coordinate multiple goroutines:

wg = Concurrency.createWaitGroup()

soul worker(id) {
    println("Worker " + id + " starting")
    Concurrency.sleep(1000)  // Simulate work
    println("Worker " + id + " done")
    wg.done()
}

// Start multiple workers
numWorkers = 5
wg.add(numWorkers)

for (i = 0; i < numWorkers; i++) {
    Concurrency.spawn(worker, i)
}

// Wait for all workers to complete
wg.wait()
println("All workers completed")

Promises

createPromise - Create promises

Create promises for async operations:

promise = Concurrency.createPromise()

// Resolve the promise
Concurrency.spawn(soul() {
    Concurrency.sleep(1000)
    promise.resolve("success")
})

// Wait for resolution
result = await promise
println(result)  // "success"

Promise.all - Wait for all promises

Wait for multiple promises to complete:

promises = []

// Create multiple async operations
for (i = 0; i < 3; i++) {
    promise = Concurrency.spawn(soul(id) {
        Concurrency.sleep(1000 + (id * 500))
        return "Task " + id + " completed"
    }, i)
    promises.push(promise)
}

// Wait for all to complete
allResults = await Concurrency.promiseAll(promises)
for (result in allResults) {
    println(result)
}

Promise.race - Wait for first promise

Wait for the first promise to complete:

promises = []

// Create multiple racing operations
for (i = 0; i < 3; i++) {
    promise = Concurrency.spawn(soul(id) {
        Concurrency.sleep(Math.random() * 2000)
        return "Task " + id + " won"
    }, i)
    promises.push(promise)
}

// Wait for first to complete
winner = await Concurrency.promiseRace(promises)
println(winner)  // "Task X won" (whichever completes first)

Timing and Timeouts

sleep - Pause execution

Sleep for a specified duration:

println("Starting...")
Concurrency.sleep(2000)  // Sleep for 2 seconds
println("After 2 seconds")

createTimeout - Timeout operations

Create timeouts for operations:

timeout = Concurrency.createTimeout(5000)  // 5 second timeout

promise = timeout.execute(soul() {
    Concurrency.sleep(3000)  // This will complete before timeout
    return "completed"
})

try {
    result = await promise
    println(result)  // "completed"
} catch (error) {
    println("Timed out: " + error.message)
}

Timeout cancellation

Cancel timeouts manually:

timeout = Concurrency.createTimeout(10000)

// Cancel the timeout
timeout.cancel()

// Check if timeout is done
if (timeout.isDone()) {
    println("Timeout was cancelled")
}

Advanced Concurrency Patterns

Worker Pool Pattern

soul createWorkerPool(numWorkers, jobs) {
    jobChannel = Concurrency.createChannel(len(jobs))
    resultChannel = Concurrency.createChannel(len(jobs))
    wg = Concurrency.createWaitGroup()
    
    // Start workers
    for (i = 0; i < numWorkers; i++) {
        wg.add(1)
        Concurrency.spawn(soul(workerId) {
            while (true) {
                try {
                    job = jobChannel.receive()
                    result = processJob(job)
                    resultChannel.send(result)
                } catch (error) {
                    if (error.message.contains("closed")) {
                        break
                    }
                }
            }
            wg.done()
        }, i)
    }
    
    // Send jobs
    for (job in jobs) {
        jobChannel.send(job)
    }
    jobChannel.close()
    
    // Collect results
    results = []
    for (i = 0; i < len(jobs); i++) {
        result = resultChannel.receive()
        results.push(result)
    }
    
    // Wait for workers to finish
    wg.wait()
    
    return results
}

soul processJob(job) {
    // Simulate work
    Concurrency.sleep(100)
    return "Processed: " + job
}

jobs = ["task1", "task2", "task3", "task4", "task5"]
results = createWorkerPool(3, jobs)
for (result in results) {
    println(result)
}

Producer-Consumer with Multiple Producers

soul multiProducerConsumer() {
    channel = Concurrency.createChannel(10)
    wg = Concurrency.createWaitGroup()
    
    // Multiple producers
    numProducers = 3
    wg.add(numProducers)
    
    for (i = 0; i < numProducers; i++) {
        Concurrency.spawn(soul(producerId) {
            for (j = 0; j < 5; j++) {
                item = "P" + producerId + "-Item" + j
                channel.send(item)
                println("Produced: " + item)
                Concurrency.sleep(100)
            }
            wg.done()
        }, i)
    }
    
    // Consumer
    Concurrency.spawn(soul() {
        itemCount = 0
        while (itemCount < numProducers * 5) {
            item = channel.receive()
            println("Consumed: " + item)
            itemCount++
        }
    })
    
    // Wait for all producers
    wg.wait()
    println("All producers finished")
}

multiProducerConsumer()

Fan-out/Fan-in Pattern

soul fanOutFanIn() {
    input = Concurrency.createChannel()
    output = Concurrency.createChannel()
    
    // Fan-out to multiple workers
    numWorkers = 3
    for (i = 0; i < numWorkers; i++) {
        Concurrency.spawn(soul(workerId) {
            while (true) {
                try {
                    job = input.receive()
                    result = "Worker" + workerId + " processed " + job
                    output.send(result)
                } catch (error) {
                    if (error.message.contains("closed")) {
                        break
                    }
                }
            }
        }, i)
    }
    
    // Send work
    Concurrency.spawn(soul() {
        for (i = 0; i < 10; i++) {
            input.send("job" + i)
        }
        input.close()
    })
    
    // Fan-in results
    for (i = 0; i < 10; i++) {
        result = output.receive()
        println(result)
    }
}

fanOutFanIn()

One-time Execution

createOnce - Ensure function runs once

Ensure a function runs only once across multiple goroutines:

once = Concurrency.createOnce()
initialized = false

soul initialize() {
    println("Initializing...")
    initialized = true
    return "initialized"
}

// Multiple goroutines try to initialize
for (i = 0; i < 5; i++) {
    Concurrency.spawn(soul() {
        result = once.do(initialize)
        println("Goroutine got: " + result)
    })
}

Concurrency.sleep(1000)
println("Initialized: " + initialized)  // Only runs once

Error Handling in Concurrent Code

Promise error handling

promise = Concurrency.spawn(soul() {
    Concurrency.sleep(1000)
    throw "Something went wrong"
})

try {
    result = await promise
    println(result)
} catch (error) {
    println("Caught error: " + error.message)
}

Channel error handling

channel = Concurrency.createChannel()

Concurrency.spawn(soul() {
    channel.send("message")
    channel.close()
})

try {
    message1 = channel.receive()
    println(message1)  // "message"
    
    message2 = channel.receive()  // This will throw
} catch (error) {
    println("Channel error: " + error.message)
}

Complete Example: Concurrent Web Scraper

soul concurrentWebScraper() {
    urls = [
        "https://example.com",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2"
    ]
    
    results = Concurrency.createChannel(len(urls))
    wg = Concurrency.createWaitGroup()
    
    soul scrapeUrl(url) {
        try {
            // Simulate web scraping
            Concurrency.sleep(1000)
            result = "Scraped: " + url
            results.send(result)
        } catch (error) {
            results.send("Error scraping " + url + ": " + error.message)
        } finally {
            wg.done()
        }
    }
    
    // Start scrapers
    for (url in urls) {
        wg.add(1)
        Concurrency.spawn(scrapeUrl, url)
    }
    
    // Collect results
    Concurrency.spawn(soul() {
        wg.wait()
        results.close()
    })
    
    // Process results
    scrapedResults = []
    while (true) {
        try {
            result = results.receive()
            scrapedResults.push(result)
        } catch (error) {
            if (error.message.contains("closed")) {
                break
            }
        }
    }
    
    return scrapedResults
}

results = concurrentWebScraper()
for (result in results) {
    println(result)
}

Best Practices

Resource cleanup: Always clean up resources (close channels, unlock mutexes)
Error handling: Handle errors in concurrent code properly
Avoid race conditions: Use proper synchronization primitives
Timeout operations: Use timeouts for potentially long-running operations
Graceful shutdown: Implement proper shutdown mechanisms

// Good - proper resource cleanup
soul properResourceManagement() {
    mutex = Concurrency.createMutex()
    
    soul safeOperation() {
        mutex.lock()
        try {
            // Do work
            return "success"
        } finally {
            mutex.unlock()
        }
    }
    
    return safeOperation()
}

// Good - timeout for operations
soul timeoutOperation() {
    timeout = Concurrency.createTimeout(5000)
    
    promise = timeout.execute(soul() {
        // Potentially long operation
        Concurrency.sleep(3000)
        return "completed"
    })
    
    try {
        result = await promise
        return result
    } catch (error) {
        println("Operation timed out")
        return null
    }
}

// Good - error handling in concurrent code
soul robustConcurrency() {
    promises = []
    
    for (i = 0; i < 3; i++) {
        promise = Concurrency.spawn(soul(id) {
            if (Math.random() > 0.5) {
                throw "Random error in task " + id
            }
            return "Task " + id + " success"
        }, i)
        promises.push(promise)
    }
    
    // Handle each promise individually
    for (promise in promises) {
        try {
            result = await promise
            println("Success: " + result)
        } catch (error) {
            println("Error: " + error.message)
        }
    }
}

The Concurrency module provides powerful tools for building concurrent and parallel applications in Soul, enabling efficient resource utilization and responsive program behavior.

CLI Usage

Modules

Syntaxes

Concurrency

Concurrency

Basic Concurrency

spawn - Execute function in goroutine

run - Fire-and-forget execution

spawn with arguments

Channels

createChannel - Create channels for communication

Channel operations

Synchronization Primitives

Mutex - Mutual exclusion

tryLock - Non-blocking lock

WaitGroup - Wait for multiple goroutines

Promises

createPromise - Create promises

Promise.all - Wait for all promises

Promise.race - Wait for first promise

Timing and Timeouts

sleep - Pause execution

createTimeout - Timeout operations

Timeout cancellation

Advanced Concurrency Patterns

Worker Pool Pattern

Producer-Consumer with Multiple Producers

Fan-out/Fan-in Pattern

One-time Execution

createOnce - Ensure function runs once

Error Handling in Concurrent Code

Promise error handling

Channel error handling

Complete Example: Concurrent Web Scraper

Best Practices

​Concurrency

​Basic Concurrency

​spawn - Execute function in goroutine

​run - Fire-and-forget execution

​spawn with arguments

​Channels

​createChannel - Create channels for communication

​Channel operations

​Synchronization Primitives

​Mutex - Mutual exclusion

​tryLock - Non-blocking lock

​WaitGroup - Wait for multiple goroutines

​Promises

​createPromise - Create promises

​Promise.all - Wait for all promises

​Promise.race - Wait for first promise

​Timing and Timeouts

​sleep - Pause execution

​createTimeout - Timeout operations

​Timeout cancellation

​Advanced Concurrency Patterns

​Worker Pool Pattern

​Producer-Consumer with Multiple Producers

​Fan-out/Fan-in Pattern

​One-time Execution

​createOnce - Ensure function runs once

​Error Handling in Concurrent Code

​Promise error handling

​Channel error handling

​Complete Example: Concurrent Web Scraper

​Best Practices

Concurrency

Basic Concurrency

spawn - Execute function in goroutine

run - Fire-and-forget execution

spawn with arguments

Channels

createChannel - Create channels for communication

Channel operations

Synchronization Primitives

Mutex - Mutual exclusion

tryLock - Non-blocking lock

WaitGroup - Wait for multiple goroutines

Promises

createPromise - Create promises

Promise.all - Wait for all promises

Promise.race - Wait for first promise

Timing and Timeouts

sleep - Pause execution

createTimeout - Timeout operations

Timeout cancellation

Advanced Concurrency Patterns

Worker Pool Pattern

Producer-Consumer with Multiple Producers

Fan-out/Fan-in Pattern

One-time Execution

createOnce - Ensure function runs once

Error Handling in Concurrent Code

Promise error handling

Channel error handling

Complete Example: Concurrent Web Scraper

Best Practices