[incubator-devlake] branch main updated: fix: stuck tasks (#5570)

[klesh]

This is an automated email from the ASF dual-hosted git repository.

klesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git


The following commit(s) were added to refs/heads/main by this push:
     new 1070214ab fix: stuck tasks (#5570)
1070214ab is described below

commit 1070214ab8eb9925b92fd7b77976b5319bb056c7
Author: Liang Zhang <liang.zh...@merico.dev>
AuthorDate: Mon Jun 26 14:04:39 2023 +0800

    fix: stuck tasks (#5570)
---
 backend/helpers/pluginhelper/api/batch_save.go | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/backend/helpers/pluginhelper/api/batch_save.go 
b/backend/helpers/pluginhelper/api/batch_save.go
index deae89b33..bf4cf06bb 100644
--- a/backend/helpers/pluginhelper/api/batch_save.go
+++ b/backend/helpers/pluginhelper/api/batch_save.go
@@ -100,19 +100,23 @@ func (c *BatchSave) Add(slot interface{}) errors.Error {
        }
        c.slots.Index(c.current).Set(reflect.ValueOf(slot))
        c.current++
-       // flush out into database if max outed
+       // flush out into database if maxed out
        if c.current == c.size {
-               return c.Flush()
+               return c.flushWithoutLocking()
        } else if c.current%100 == 0 {
                c.log.Debug("batch save current: %d", c.current)
        }
        return nil
 }
 
-// Flush save cached records into database
+// Flush save cached records into database, even if cache is not maxed out
 func (c *BatchSave) Flush() errors.Error {
        c.mutex.Lock()
        defer c.mutex.Unlock()
+       return c.flushWithoutLocking()
+}
+
+func (c *BatchSave) flushWithoutLocking() errors.Error {
        if c.current == 0 {
                return nil
        }
@@ -130,10 +134,12 @@ func (c *BatchSave) Flush() errors.Error {
        return nil
 }
 
-// Close would flash the cache and release resources
+// Close would flush the cache and release resources
 func (c *BatchSave) Close() errors.Error {
+       c.mutex.Lock()
+       defer c.mutex.Unlock()
        if c.current > 0 {
-               return c.Flush()
+               return c.flushWithoutLocking()
        }
        return nil
 }