Implement sync

src/internal/task/queue.go

@@ -68,8 +68,8 @@ func (s *Stack) Pop() *Task {
 	t := s.top
 	if t != nil {
 		s.top = t.Next
+		t.Next = nil
 	}
-	t.Next = nil
 	return t
 }
 

src/sync/cond.go

@@ -0,0 +1,80 @@
+package sync
+
+import "internal/task"
+
+type Cond struct {
+	L Locker
+
+	unlocking *earlySignal
+	blocked   task.Stack
+}
+
+// earlySignal is a type used to implement a stack for signalling waiters while they are unlocking.
+type earlySignal struct {
+	next *earlySignal
+
+	signaled bool
+}
+
+func (c *Cond) trySignal() bool {
+	// Pop a blocked task off of the stack, and schedule it if applicable.
+	t := c.blocked.Pop()
+	if t != nil {
+		scheduleTask(t)
+		return true
+	}
+
+	// If there any tasks which are currently unlocking, signal one.
+	if c.unlocking != nil {
+		c.unlocking.signaled = true
+		c.unlocking = c.unlocking.next
+		return true
+	}
+
+	// There was nothing to signal.
+	return false
+}
+
+func (c *Cond) Signal() {
+	c.trySignal()
+}
+
+func (c *Cond) Broadcast() {
+	// Signal everything.
+	for c.trySignal() {
+	}
+}
+
+func (c *Cond) Wait() {
+	// Add an earlySignal frame to the stack so we can be signalled while unlocking.
+	early := earlySignal{
+		next: c.unlocking,
+	}
+	c.unlocking = &early
+
+	// Temporarily unlock L.
+	c.L.Unlock()
+
+	// Re-acquire the lock before returning.
+	defer c.L.Lock()
+
+	// If we were signaled while unlocking, immediately complete.
+	if early.signaled {
+		return
+	}
+
+	// Remove the earlySignal frame.
+	prev := c.unlocking
+	for prev != nil && prev.next != &early {
+		prev = prev.next
+	}
+	if prev != nil {
+		prev.next = early.next
+	} else {
+		c.unlocking = early.next
+	}
+
+	// Wait for a signal.
+	c.blocked.Push(task.Current())
+	task.Pause()
+}

src/sync/mutex.go

@@ -1,24 +1,43 @@
 package sync
 
+import (
+	"internal/task"
+	_ "unsafe"
+)
+
 // These mutexes assume there is only one thread of operation: no goroutines,
 // interrupts or anything else.
 
 type Mutex struct {
-	locked bool
+	locked  bool
+	blocked task.Stack
 }
 
+//go:linkname scheduleTask runtime.runqueuePushBack
+func scheduleTask(*task.Task)
+
 func (m *Mutex) Lock() {
 	if m.locked {
-		panic("todo: block on locked mutex")
+		// Push self onto stack of blocked tasks, and wait to be resumed.
+		m.blocked.Push(task.Current())
+		task.Pause()
+		return
 	}
+
 	m.locked = true
 }
 
 func (m *Mutex) Unlock() {
 	if !m.locked {
 		panic("sync: unlock of unlocked Mutex")
 	}
-	m.locked = false
+
+	// Wake up a blocked task, if applicable.
+	if t := m.blocked.Pop(); t != nil {
+		scheduleTask(t)
+	} else {
+		m.locked = false
+	}
 }
 
 type RWMutex struct {
@@ -50,3 +69,8 @@ func (rw *RWMutex) RUnlock() {
 		rw.m.Unlock()
 	}
 }
+
+type Locker interface {
+	Lock()
+	Unlock()
+}

src/sync/waitgroup.go

@@ -0,0 +1,54 @@
+package sync
+
+import "internal/task"
+
+type WaitGroup struct {
+	counter uint
+	waiters task.Stack
+}
+
+func (wg *WaitGroup) Add(delta int) {
+	if delta > 0 {
+		// Check for overflow.
+		if uint(delta) > (^uint(0))-wg.counter {
+			panic("sync: WaitGroup counter overflowed")
+		}
+
+		// Add to the counter.
+		wg.counter += uint(delta)
+	} else {
+		// Check for underflow.
+		if uint(-delta) > wg.counter {
+			panic("sync: negative WaitGroup counter")
+		}
+
+		// Subtract from the counter.
+		wg.counter -= uint(-delta)
+
+		// If the counter is zero, everything is done and the waiters should be resumed.
+		// This code assumes that the waiters cannot wake up until after this function returns.
+		// In the current implementation, this is always correct.
+		if wg.counter == 0 {
+			for t := wg.waiters.Pop(); t != nil; t = wg.waiters.Pop() {
+				scheduleTask(t)
+			}
+		}
+	}
+}
+
+func (wg *WaitGroup) Done() {
+	wg.Add(-1)
+}
+
+func (wg *WaitGroup) Wait() {
+	if wg.counter == 0 {
+		// Everything already finished.
+		return
+	}
+
+	// Push the current goroutine onto the waiter stack.
+	wg.waiters.Push(task.Current())
+
+	// Pause until the waiters are awoken by Add/Done.
+	task.Pause()
+}