A queue stores elements in an ordered sequence and removes them according to a policy, typically first in, first out (FIFO). In concurrent programs, it buffers messages or work between producers and consumers. section 7.2.1 section 10.2.3
A concurrent queue implementation needs:
- A shared buffer, such as a Go slice, array-backed queue, or linked list. section 7.2.1
- A mutex to prevent simultaneous enqueue and dequeue operations from interfering. section 7.2.1
- Access control that blocks receivers when the queue is empty and senders when it is full. section 7.2.1
A queue behind a worker pool
An HTTP server can create several goroutines that consume connections from one Go channel. Idle workers receive queued connections; when all workers are busy, the queue grows. A bounded queue can reject additional connections with a busy response instead of accepting unlimited work. section 10.2.3
Listing 7.14 Type struct for the custom channel implementation
package listing7_14 import ( "container/list" "github.com/cutajarj/ConcurrentProgrammingWithGo/chapter5/listing5.16" "sync" ) type Channel[M any] struct { capacitySema *listing5_16.Semaphore #1 sizeSema *listing5_16.Semaphore #2 mutex sync.Mutex #3 buffer *list.List #4 } #1 Capacity semaphore to block sender when the buffer is full #2 Buffer size semaphore to block the receiver when the buffer is empty #3 Mutex protecting our shared list data structure #4 Linked list to be used as a queue data structure
Inspect a custom channel structure combining a linked-list queue, mutex, capacity semaphore, and buffer-size semaphore. section 7.2.1
Listing 10.12 Using select’s default case to limit the load on the server
package main import ( "fmt" "github.com/cutajarj/ConcurrentProgrammingWithGo/chapter10/listing10.9" "net" ) func main() { incomingConnections := make(chan net.Conn) listing10_9.StartHttpWorkers(3, incomingConnections) server, _ := net.Listen("tcp", "localhost:8080") defer server.Close() for { conn, _ := server.Accept() select { case incomingConnections <- conn: default: #1 fmt.Println("Server is busy") conn.Write([]byte("HTTP/1.1 429 Too Many Requests\r\n\r\n" + "<html>Busy</html>\n") #2 conn.Close() #3 } } } #1 When no goroutines are consuming from the work queue, the default case triggers. #2 Returns “busy” message to client #3 Closes client connection
Inspect a nonblocking send that limits server load when no worker is ready to consume queued work. section 10.2.3
