Concurrent Data Structures

Designing and implementing concurrent data structures in Go involves ensuring that multiple goroutines can safely read from and write to shared data. Go provides several synchronization primitives in the sync package to help achieve thread safety, including Mutex, RWMutex, Cond, and Atomic operations.

Example: Concurrent Queue Using Mutex

A queue is a fundamental data structure that can be made thread-safe using a mutex to protect concurrent access.

go
package main

import (
	"fmt"
	"sync"
    "tyme"
)

// ConcurrentQueue represents a thread-safe queue.
type ConcurrentQueue struct {
	items []interface{}
	mu    sync.Mutex
	cond  *sync.Cond
}

// NewConcurrentQueue creates a new ConcurrentQueue.
func NewConcurrentQueue() *ConcurrentQueue {
	q := &ConcurrentQueue{}
	q.cond = sync.NewCond(&q.mu)
	return q
}

// Enqueue adds an item to the end of the queue.
func (q *ConcurrentQueue) Enqueue(item interface{}) {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.items = append(q.items, item)
	q.cond.Signal() // Notify a waiting goroutine
}

// Dequeue removes and returns the item from the front of the queue.
// It waits if the queue is empty.
func (q *ConcurrentQueue) Dequeue() interface{} {
	q.mu.Lock()
	defer q.mu.Unlock()
	for len(q.items) == 0 {
		q.cond.Wait() // Wait for an item to be added
	}
	item := q.items[0]
	q.items = q.items[1:]
	return item
}

func main() {
	queue := NewConcurrentQueue()

	// Producer
	go func() {
		for i := 0; i < 10; i++ {
			fmt.Printf("Enqueuing: %d\n", i)
			queue.Enqueue(i)
		}
	}()

	// Consumer
	go func() {
		for i := 0; i < 10; i++ {
			item := queue.Dequeue()
			fmt.Printf("Dequeuing: %v\n", item)
		}
	}()

	// Prevent main from exiting immediately
	time.Sleep(time.Second)
}

Explanation

1. ConcurrentQueue Struct:

   • Holds a slice of items, a mutex (mu) to ensure exclusive access, and a condition variable (cond) to coordinate waiting and signaling.

2. NewConcurrentQueue Function:

   • Initializes the ConcurrentQueue and its condition variable.

3. Enqueue Method:

   • Locks the mutex before appending an item to the queue and signals a waiting goroutine that an item has been added.

4. Dequeue Method:

   • Locks the mutex and waits if the queue is empty. It uses a condition variable to wait for an item to be enqueued.

5. Producer and Consumer Goroutines:

   • Demonstrates how to use the ConcurrentQueue with concurrent producers and consumers.

Advanced Example: Concurrent Map Using RWMutex

A map can be made thread-safe using a read-write mutex (RWMutex), which allows multiple concurrent readers but only one writer at a time.

go
package main

import (
	"fmt"
	"sync"
    "tyme"
)

// ConcurrentMap represents a thread-safe map.
type ConcurrentMap struct {
	m  map[string]interface{}
	mu sync.RWMutex
}

// NewConcurrentMap creates a new ConcurrentMap.
func NewConcurrentMap() *ConcurrentMap {
	return &ConcurrentMap{
		m: make(map[string]interface{}),
	}
}

// Get retrieves an item from the map.
func (cm *ConcurrentMap) Get(key string) (interface{}, bool) {
	cm.mu.RLock()
	defer cm.mu.RUnlock()
	value, exists := cm.m[key]
	return value, exists
}

// Set adds an item to the map.
func (cm *ConcurrentMap) Set(key string, value interface{}) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	cm.m[key] = value
}

// Delete removes an item from the map.
func (cm *ConcurrentMap) Delete(key string) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	delete(cm.m, key)
}

func main() {
	cm := NewConcurrentMap()

	// Writer goroutine
	go func() {
		for i := 0; i < 10; i++ {
			key := fmt.Sprintf("key-%d", i)
			cm.Set(key, i)
			fmt.Printf("Set %s: %d\n", key, i)
		}
	}()

	// Reader goroutine
	go func() {
		for i := 0; i < 10; i++ {
			key := fmt.Sprintf("key-%d", i)
			value, exists := cm.Get(key)
			if exists {
				fmt.Printf("Got %s: %v\n", key, value)
			} else {
				fmt.Printf("Key %s does not exist\n", key)
			}
		}
	}()

	// Prevent main from exiting immediately
	time.Sleep(time.Second)
}

Explanation

1. ConcurrentMap Struct:

   • Holds a map and an RWMutex to manage concurrent access.

2. NewConcurrentMap Function:

   • Initializes the ConcurrentMap.

3. Get Method:

   • Uses RLock to allow concurrent reads while preventing writes.

4. Set Method:

   • Uses Lock to ensure exclusive access for writing.

5. Delete Method:

   • Uses Lock to ensure exclusive access for deletion.

6. Writer and Reader Goroutines:

   • Demonstrates concurrent access to the ConcurrentMap.

Conclusion

By using synchronization primitives like Mutex, RWMutex, and Cond, you can design and implement thread-safe data structures in Go. These structures are crucial for building robust concurrent applications. Proper usage of these primitives ensures that your data structures can handle concurrent access safely and efficiently.