feat: build cancel support added with deadlock fixation 🎉 (#184)

Author: tanmoysrt

container_manager/image.go

@@ -2,6 +2,7 @@ package containermanger
 
 import (
 	"bufio"
+	"context"
 	"errors"
 	"os"
 
@@ -15,6 +16,16 @@ It takes the Dockerfile content as a string, a map of build arguments, the path
 It returns a scanner to read the build logs and an error if any.
 */
 func (m Manager) CreateImage(dockerfile string, buildargs map[string]string, codepath string, imagename string) (*bufio.Scanner, error) {
+	return m.CreateImageWithContext(m.ctx, dockerfile, buildargs, codepath, imagename)
+}
+
+/*
+CreateImageWithContext builds a Docker image from a Dockerfile and returns a scanner to read the build logs.
+It takes the Dockerfile content as a string, a map of build arguments, the path to the code directory, and the name of the image to be built.
+It returns a scanner to read the build logs and an error if any.
+It takes a context.Context as an additional argument.
+*/
+func (m Manager) CreateImageWithContext(ctx context.Context, dockerfile string, buildargs map[string]string, codepath string, imagename string) (*bufio.Scanner, error) {
 	// Move the dockerfile to the codepath
 	err := os.WriteFile(codepath+"/Dockerfile", []byte(dockerfile), 0777)
 	if err != nil {
@@ -35,7 +46,7 @@ func (m Manager) CreateImage(dockerfile string, buildargs map[string]string, cod
 		return nil, errors.New("failed to tar the codepath")
 	}
 	// Build the image
-	response, err := m.client.ImageBuild(m.ctx, tar, types.ImageBuildOptions{
+	response, err := m.client.ImageBuild(ctx, tar, types.ImageBuildOptions{
 		Dockerfile: "Dockerfile",
 		Remove:     true,
 		Tags:       []string{imagename},

dontdo.md

@@ -2,4 +2,5 @@
 
 - `addDeploymentLog` function is safe to use without transaction at the time of using transaction in another thread.
 - Except that above one, don't run any function without transaction at the time of using transaction.
-- Avoid to use `dbWithoutTx` until unless required. Use carefully as it can lead to database deadlock condition.
+- Avoid to use `dbWithoutTx` until unless required. Use carefully as it can lead to database deadlock condition.
+- Don't use just Read Lock while closing channel, do RW lock

pubsub/local_pubsub.go

@@ -3,6 +3,7 @@ package pubsub
 import (
 	"errors"
 	"github.com/google/uuid"
+	"github.com/hashicorp/go-set"
 	"log"
 	"sync"
 )
@@ -11,15 +12,17 @@ func (l *localPubSub) CreateTopic(topic string) error {
 	if l.closed {
 		return errors.New("pubsub client is closed")
 	}
-	if l.topics.Contains(topic) {
+	l.mutex.RLock()
+	isContains := l.topics.Contains(topic)
+	l.mutex.RUnlock()
+	if isContains {
 		return nil
 	} else {
-		// lock
 		l.mutex.Lock()
-		defer l.mutex.Unlock()
 		// insert
 		l.topics.Insert(topic)
 		l.subscriptions[topic] = make(map[string]localPubSubSubscription)
+		l.mutex.Unlock()
 		return nil
 	}
 }
@@ -28,56 +31,45 @@ func (l *localPubSub) RemoveTopic(topic string) error {
 	if l.closed {
 		return errors.New("pubsub client is closed")
 	}
-	// lock
-	l.mutex.Lock()
-	defer l.mutex.Unlock()
+	l.mutex.RLock()
+	isContains := l.topics.Contains(topic)
+	l.mutex.RUnlock()
 	// check if topic exists
-	if !l.topics.Contains(topic) {
+	if !isContains {
 		return nil
 	} else {
-		// close and remove all subscribers
-		for subscriptionRecord := range l.subscriptions[topic] {
-			l.cancelSubscriptions(topic, subscriptionRecord)
+		l.mutex.Lock()
+		// close all subscribers
+		subscriptionRecords := l.subscriptions[topic]
+		for _, subscription := range subscriptionRecords {
+			m := subscription.Mutex
+			m.Lock()
+			close(subscription.Channel)
+			m.Unlock()
 		}
-		// delete subscribers
+		// delete topic
 		delete(l.subscriptions, topic)
+		l.mutex.Unlock()
 	}
 	return nil
 }
 
-func (l *localPubSub) cancelSubscriptions(topic string, subscriptionId string) {
-	if l.closed {
-		return
-	}
-	// verify topic exists with ok
-	if _, ok := l.subscriptions[topic]; !ok {
-		return
-	}
-	// verify subscription exists
-	if _, ok := l.subscriptions[topic][subscriptionId]; !ok {
-		return
-	}
-	// lock
-	mutex := l.subscriptions[topic][subscriptionId].Mutex
-	mutex.Lock()
-	defer mutex.Unlock()
-	// close channel
-	close(l.subscriptions[topic][subscriptionId].Channel)
-	// delete subscription
-	delete(l.subscriptions[topic], subscriptionId)
-}
-
 // Subscribe returns a subscription id and a channel to listen to
 func (l *localPubSub) Subscribe(topic string) (string, <-chan string, error) {
 	if l.closed {
 		return "", nil, errors.New("pubsub client is closed")
 	}
 	// lock
-	l.mutex.Lock()
-	defer l.mutex.Unlock()
+	l.mutex.RLock()
+	isContains := l.topics.Contains(topic)
+	l.mutex.RUnlock()
 	// check if topic exists
-	if !l.topics.Contains(topic) {
-		return "", nil, errors.New("topic does not exist")
+	if !isContains {
+		l.mutex.Lock()
+		// insert topic
+		l.topics.Insert(topic)
+		l.subscriptions[topic] = make(map[string]localPubSubSubscription)
+		l.mutex.Unlock()
 	}
 	// create a new subscription id
 	subscriptionId := topic + "_" + uuid.NewString()
@@ -88,8 +80,12 @@ func (l *localPubSub) Subscribe(topic string) (string, <-chan string, error) {
 		Mutex:   &sync.RWMutex{},
 		Channel: channel,
 	}
+	// critical section
+	l.mutex.Lock()
 	// add subscription record to subscriptions
 	l.subscriptions[topic][subscriptionId] = subscriptionRecord
+	// unlock
+	l.mutex.Unlock()
 	// return subscription id and channel
 	return subscriptionId, channel, nil
 }
@@ -100,29 +96,32 @@ func (l *localPubSub) Unsubscribe(topic string, subscriptionId string) error {
 		return errors.New("pubsub client is closed")
 	}
 	// lock main mutex
-	l.mutex.Lock()
-	defer l.mutex.Unlock()
+	l.mutex.RLock()
+	isContains := l.topics.Contains(topic)
+	l.mutex.RUnlock()
 	// check if topic exists
-	if !l.topics.Contains(topic) {
-		return errors.New("topic does not exist")
+	if !isContains {
+		return nil
 	}
 	// check if subscription exists
+	l.mutex.RLock()
 	if _, ok := l.subscriptions[topic][subscriptionId]; !ok {
-		return errors.New("subscription does not exist")
+		l.mutex.RUnlock()
+		return nil
 	}
 	// fetch subscription record
 	subscriptionRecord := l.subscriptions[topic][subscriptionId]
+	l.mutex.RUnlock()
 	// lock
 	mutex := subscriptionRecord.Mutex
 	mutex.Lock()
-	defer mutex.Unlock()
 	// cleanup channel
-	if _, ok := <-subscriptionRecord.Channel; ok {
-		<-subscriptionRecord.Channel
-		close(subscriptionRecord.Channel)
-	}
+	close(subscriptionRecord.Channel)
+	mutex.Unlock()
+	l.mutex.Lock()
 	// delete subscription
 	delete(l.subscriptions[topic], subscriptionId)
+	l.mutex.Unlock()
 	return nil
 }
 
@@ -131,29 +130,33 @@ func (l *localPubSub) Publish(topic string, data string) error {
 	if l.closed {
 		return errors.New("pubsub client is closed")
 	}
-	// lock main mutex
-	l.mutex.Lock()
-	defer l.mutex.Unlock()
+	l.mutex.RLock()
+	isContains := l.topics.Contains(topic)
+	l.mutex.RUnlock()
 	// check if topic exists
-	if !l.topics.Contains(topic) {
+	if !isContains {
+		l.mutex.Lock()
 		// insert topic
 		l.topics.Insert(topic)
 		l.subscriptions[topic] = make(map[string]localPubSubSubscription)
+		l.mutex.Unlock()
 	}
 	// fetch all subscriptions
+	l.mutex.RLock()
 	subscriptions := l.subscriptions[topic]
+	l.mutex.RUnlock()
 	// iterate over all subscriptions
 	for _, subscriptionRecord := range subscriptions {
 		// lock subscription mutex
 		mutex := subscriptionRecord.Mutex
 		mutex.Lock()
+		channel := subscriptionRecord.Channel
 		// clear channel if full
-		if len(subscriptionRecord.Channel) == cap(subscriptionRecord.Channel) {
-			<-subscriptionRecord.Channel
+		if len(channel) == cap(channel) {
+			<-channel
 		}
 		// send data
-		subscriptionRecord.Channel <- data
-		// unlock subscription mutex
+		channel <- data
 		mutex.Unlock()
 	}
 	return nil
@@ -177,11 +180,17 @@ func (l *localPubSub) Close() error {
 	}
 	// set closed to true
 	l.closed = true
-	// close and remove all subscribers
+	// close
 	for topic := range l.subscriptions {
-		for subscriptionRecord := range l.subscriptions[topic] {
-			l.cancelSubscriptions(topic, subscriptionRecord)
+		for _, subscription := range l.subscriptions[topic] {
+			m := subscription.Mutex
+			m.Lock()
+			close(subscription.Channel)
+			m.Unlock()
 		}
 	}
+	// remove all topics
+	l.topics = set.New[string](0)
+	l.subscriptions = make(map[string]map[string]localPubSubSubscription)
 	return nil
 }