You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@devlake.apache.org by kl...@apache.org on 2024/02/01 04:21:51 UTC
(incubator-devlake) branch kw-6899-cronconfig-deadlock created (now ca12670e0)
This is an automated email from the ASF dual-hosted git repository.
klesh pushed a change to branch kw-6899-cronconfig-deadlock
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
at ca12670e0 fix: triggering blueprint concurrently might lead to deadlock
This branch includes the following new commits:
new ca12670e0 fix: triggering blueprint concurrently might lead to deadlock
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(incubator-devlake) 01/01: fix: triggering blueprint concurrently might lead to deadlock
Posted by kl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
klesh pushed a commit to branch kw-6899-cronconfig-deadlock
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
commit ca12670e0e3ccae5f23cd79f353b686885fb1c1d
Author: Klesh Wong <zh...@merico.dev>
AuthorDate: Thu Feb 1 12:18:53 2024 +0800
fix: triggering blueprint concurrently might lead to deadlock
---
backend/server/services/pipeline.go | 13 ++++++++-----
backend/server/services/pipeline_helper.go | 3 ++-
backend/server/services/task.go | 5 ++---
3 files changed, 12 insertions(+), 9 deletions(-)
diff --git a/backend/server/services/pipeline.go b/backend/server/services/pipeline.go
index a72b0b48f..ab72c97d5 100644
--- a/backend/server/services/pipeline.go
+++ b/backend/server/services/pipeline.go
@@ -20,7 +20,6 @@ package services
import (
"context"
"fmt"
- "github.com/spf13/cast"
"net/url"
"os"
"path/filepath"
@@ -28,6 +27,8 @@ import (
"sync"
"time"
+ "github.com/spf13/cast"
+
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/models"
@@ -413,7 +414,10 @@ func RerunPipeline(pipelineId uint64, task *models.Task) (tasks []*models.Task,
txHelper := dbhelper.NewTxHelper(basicRes, &err)
tx := txHelper.Begin()
defer txHelper.End()
- err = txHelper.LockTablesTimeout(2*time.Second, dal.LockTables{{Table: "_devlake_pipelines", Exclusive: true}})
+ err = txHelper.LockTablesTimeout(2*time.Second, dal.LockTables{
+ {Table: "_devlake_pipelines", Exclusive: true},
+ {Table: "_devlake_tasks", Exclusive: true},
+ })
if err != nil {
err = errors.BadInput.Wrap(err, "failed to lock pipeline table, is there any pending pipeline or deletion?")
return
@@ -458,7 +462,6 @@ func RerunPipeline(pipelineId uint64, task *models.Task) (tasks []*models.Task,
}
// create new tasks
- // TODO: this is better to be wrapped inside a transaction
rerunTasks := []*models.Task{}
for _, t := range failedTasks {
// mark previous task failed
@@ -468,7 +471,7 @@ func RerunPipeline(pipelineId uint64, task *models.Task) (tasks []*models.Task,
return nil, err
}
// create new task
- rerunTask, err := CreateTask(&models.NewTask{
+ rerunTask, err := createTask(&models.NewTask{
PipelineTask: &models.PipelineTask{
Plugin: t.Plugin,
Subtasks: t.Subtasks,
@@ -478,7 +481,7 @@ func RerunPipeline(pipelineId uint64, task *models.Task) (tasks []*models.Task,
PipelineRow: t.PipelineRow,
PipelineCol: t.PipelineCol,
IsRerun: true,
- })
+ }, tx)
if err != nil {
return nil, err
}
diff --git a/backend/server/services/pipeline_helper.go b/backend/server/services/pipeline_helper.go
index 11c00e94d..eec6a8c34 100644
--- a/backend/server/services/pipeline_helper.go
+++ b/backend/server/services/pipeline_helper.go
@@ -37,6 +37,7 @@ func CreateDbPipeline(newPipeline *models.NewPipeline) (pipeline *models.Pipelin
dal.LockTables{
{Table: "_devlake_pipelines", Exclusive: true},
{Table: "_devlake_pipeline_labels", Exclusive: true},
+ {Table: "_devlake_tasks", Exclusive: true},
},
))
if err != nil {
@@ -91,7 +92,7 @@ func CreateDbPipeline(newPipeline *models.NewPipeline) (pipeline *models.Pipelin
PipelineRow: i + 1,
PipelineCol: j + 1,
}
- _ = errors.Must1(CreateTask(newTask))
+ _ = errors.Must1(createTask(newTask, tx))
// sync task state back to pipeline
dbPipeline.TotalTasks += 1
}
diff --git a/backend/server/services/task.go b/backend/server/services/task.go
index e3e9a1fb6..9c4922e17 100644
--- a/backend/server/services/task.go
+++ b/backend/server/services/task.go
@@ -40,8 +40,7 @@ type TaskQuery struct {
Pending int `form:"pending"`
}
-// CreateTask creates a new task
-func CreateTask(newTask *models.NewTask) (*models.Task, errors.Error) {
+func createTask(newTask *models.NewTask, tx dal.Transaction) (*models.Task, errors.Error) {
task := &models.Task{
Plugin: newTask.Plugin,
Subtasks: newTask.Subtasks,
@@ -55,7 +54,7 @@ func CreateTask(newTask *models.NewTask) (*models.Task, errors.Error) {
if newTask.IsRerun {
task.Status = models.TASK_RERUN
}
- err := db.Create(task)
+ err := tx.Create(task)
if err != nil {
taskLog.Error(err, "save task failed")
return nil, errors.Internal.Wrap(err, "save task failed")