bugfix: race conditions
This commit is contained in:
parent
221c85685f
commit
c790a8c482
|
@ -91,6 +91,8 @@ func TestCallback(t *testing.T) {
|
||||||
|
|
||||||
// TODO use callback or chan chan to avoid sleeps?
|
// TODO use callback or chan chan to avoid sleeps?
|
||||||
time.Sleep(debounceDelay)
|
time.Sleep(debounceDelay)
|
||||||
|
t.Log("sleep so job can finish")
|
||||||
|
time.Sleep(jobDelay)
|
||||||
time.Sleep(jobDelay)
|
time.Sleep(jobDelay)
|
||||||
|
|
||||||
// TODO test that the API gives this back to us
|
// TODO test that the API gives this back to us
|
||||||
|
|
|
@ -108,7 +108,7 @@ func Run(runOpts *options.ServerConfig) {
|
||||||
run(hook, runOpts)
|
run(hook, runOpts)
|
||||||
case activeID := <-deathRow:
|
case activeID := <-deathRow:
|
||||||
//log.Printf("[%s] done", activeID)
|
//log.Printf("[%s] done", activeID)
|
||||||
remove(activeID /*, false*/)
|
remove(runOpts, activeID /*, false*/)
|
||||||
case promotion := <-Promotions:
|
case promotion := <-Promotions:
|
||||||
log.Printf("[%s] promoting to %s", promotion.GitRef.GetRefID(), promotion.PromoteTo)
|
log.Printf("[%s] promoting to %s", promotion.GitRef.GetRefID(), promotion.PromoteTo)
|
||||||
promote(webhooks.New(*promotion.GitRef), promotion.PromoteTo, runOpts)
|
promote(webhooks.New(*promotion.GitRef), promotion.PromoteTo, runOpts)
|
||||||
|
@ -464,30 +464,6 @@ func run(curHook *webhooks.Ref, runOpts *options.ServerConfig) {
|
||||||
_ = txtFile.Close()
|
_ = txtFile.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO move to deathRow only?
|
|
||||||
updateExitStatus(j)
|
|
||||||
|
|
||||||
// Switch ID to the more specific RevID
|
|
||||||
j.ID = string(j.GitRef.GetRevID())
|
|
||||||
// replace the text log with a json log
|
|
||||||
if jsonFile, err := getJobFile(runOpts.LogDir, j.GitRef, ".json"); nil != err {
|
|
||||||
// jsonFile.Name() should be the full path
|
|
||||||
log.Printf("[warn] could not create log file '%s': %v", runOpts.LogDir, err)
|
|
||||||
} else {
|
|
||||||
enc := json.NewEncoder(jsonFile)
|
|
||||||
enc.SetIndent("", " ")
|
|
||||||
if err := enc.Encode(j); nil != err {
|
|
||||||
log.Printf("[warn] could not encode json log '%s': %v", jsonFile.Name(), err)
|
|
||||||
} else {
|
|
||||||
logdir, logname, _ := getJobFilePath(runOpts.LogDir, j.GitRef, ".log")
|
|
||||||
_ = os.Remove(filepath.Join(logdir, logname))
|
|
||||||
}
|
|
||||||
_ = jsonFile.Close()
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO move to deathRow only?
|
|
||||||
j.Logs = []Log{}
|
|
||||||
|
|
||||||
// this will completely clear the finished job
|
// this will completely clear the finished job
|
||||||
deathRow <- pendingID
|
deathRow <- pendingID
|
||||||
|
|
||||||
|
@ -542,7 +518,7 @@ func getEnvs(addr, activeID string, repoList string, hook *webhooks.Ref) []strin
|
||||||
}
|
}
|
||||||
|
|
||||||
// Remove kills the job and moves it to recents
|
// Remove kills the job and moves it to recents
|
||||||
func remove(activeID webhooks.RefID /*, nokill bool*/) {
|
func remove(runOpts *options.ServerConfig, activeID webhooks.RefID /*, nokill bool*/) {
|
||||||
// Encapsulate the whole transaction
|
// Encapsulate the whole transaction
|
||||||
jobsTimersMux.Lock()
|
jobsTimersMux.Lock()
|
||||||
defer jobsTimersMux.Unlock()
|
defer jobsTimersMux.Unlock()
|
||||||
|
@ -555,17 +531,6 @@ func remove(activeID webhooks.RefID /*, nokill bool*/) {
|
||||||
job := value.(*Job)
|
job := value.(*Job)
|
||||||
Actives.Delete(activeID)
|
Actives.Delete(activeID)
|
||||||
|
|
||||||
// transition to RevID for non-active, non-pending jobs
|
|
||||||
job.ID = string(job.GitRef.GetRevID())
|
|
||||||
Recents.Store(job.GitRef.GetRevID(), job)
|
|
||||||
|
|
||||||
updateExitStatus(job)
|
|
||||||
|
|
||||||
// JSON should have been written to disk by this point
|
|
||||||
job.Logs = []Log{}
|
|
||||||
}
|
|
||||||
|
|
||||||
func updateExitStatus(job *Job) {
|
|
||||||
if nil == job.cmd.ProcessState {
|
if nil == job.cmd.ProcessState {
|
||||||
// is not yet finished
|
// is not yet finished
|
||||||
if nil != job.cmd.Process {
|
if nil != job.cmd.Process {
|
||||||
|
@ -582,6 +547,29 @@ func updateExitStatus(job *Job) {
|
||||||
}
|
}
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
job.EndedAt = &now
|
job.EndedAt = &now
|
||||||
|
|
||||||
|
// Switch ID to the more specific RevID
|
||||||
|
job.ID = string(job.GitRef.GetRevID())
|
||||||
|
// replace the text log with a json log
|
||||||
|
if jsonFile, err := getJobFile(runOpts.LogDir, job.GitRef, ".json"); nil != err {
|
||||||
|
// jsonFile.Name() should be the full path
|
||||||
|
log.Printf("[warn] could not create log file '%s': %v", runOpts.LogDir, err)
|
||||||
|
} else {
|
||||||
|
enc := json.NewEncoder(jsonFile)
|
||||||
|
enc.SetIndent("", " ")
|
||||||
|
if err := enc.Encode(job); nil != err {
|
||||||
|
log.Printf("[warn] could not encode json log '%s': %v", jsonFile.Name(), err)
|
||||||
|
} else {
|
||||||
|
logdir, logname, _ := getJobFilePath(runOpts.LogDir, job.GitRef, ".log")
|
||||||
|
_ = os.Remove(filepath.Join(logdir, logname))
|
||||||
|
}
|
||||||
|
_ = jsonFile.Close()
|
||||||
|
}
|
||||||
|
job.Logs = []Log{}
|
||||||
|
|
||||||
|
// transition to RevID for non-active, non-pending jobs
|
||||||
|
job.ID = string(job.GitRef.GetRevID())
|
||||||
|
Recents.Store(job.GitRef.GetRevID(), job)
|
||||||
}
|
}
|
||||||
|
|
||||||
func expire(runOpts *options.ServerConfig) {
|
func expire(runOpts *options.ServerConfig) {
|
||||||
|
|
|
@ -89,7 +89,7 @@ func TestDebounce(t *testing.T) {
|
||||||
Repo: "repo",
|
Repo: "repo",
|
||||||
})
|
})
|
||||||
|
|
||||||
//t.Log("sleep so job can debounce and start")
|
t.Log("sleep so job can debounce and start")
|
||||||
time.Sleep(debounceDelay)
|
time.Sleep(debounceDelay)
|
||||||
|
|
||||||
var jobMatch *Job
|
var jobMatch *Job
|
||||||
|
@ -144,10 +144,10 @@ func TestDebounce(t *testing.T) {
|
||||||
Repo: "repo",
|
Repo: "repo",
|
||||||
})
|
})
|
||||||
|
|
||||||
//t.Log("sleep so 1st job can finish")
|
t.Log("sleep so 1st job can finish")
|
||||||
time.Sleep(jobDelay)
|
time.Sleep(jobDelay)
|
||||||
time.Sleep(jobDelay)
|
time.Sleep(jobDelay)
|
||||||
//t.Log("sleep so backlog can debounce")
|
t.Log("sleep so backlog can debounce")
|
||||||
time.Sleep(debounceDelay)
|
time.Sleep(debounceDelay)
|
||||||
|
|
||||||
//var j *Job
|
//var j *Job
|
||||||
|
@ -176,10 +176,10 @@ func TestDebounce(t *testing.T) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
//t.Log("sleep so 2nd job can finish")
|
t.Log("sleep so 2nd job can finish")
|
||||||
time.Sleep(jobDelay)
|
time.Sleep(jobDelay)
|
||||||
|
|
||||||
//t.Log("sleep to ensure no more backlogs exist")
|
t.Log("sleep to ensure no more backlogs exist")
|
||||||
time.Sleep(jobDelay)
|
time.Sleep(jobDelay)
|
||||||
time.Sleep(debounceDelay)
|
time.Sleep(debounceDelay)
|
||||||
time.Sleep(debounceDelay)
|
time.Sleep(debounceDelay)
|
||||||
|
@ -225,9 +225,11 @@ func TestRecents(t *testing.T) {
|
||||||
}
|
}
|
||||||
Debounce(hook)
|
Debounce(hook)
|
||||||
|
|
||||||
//t.Log("sleep so job can debounce and start")
|
t.Log("sleep so job can debounce and start")
|
||||||
time.Sleep(debounceDelay)
|
time.Sleep(debounceDelay)
|
||||||
time.Sleep(jobDelay)
|
time.Sleep(jobDelay)
|
||||||
|
t.Log("sleep so job can finish")
|
||||||
|
time.Sleep(jobDelay)
|
||||||
|
|
||||||
urlRefID := webhooks.URLSafeGitID(
|
urlRefID := webhooks.URLSafeGitID(
|
||||||
base64.RawURLEncoding.EncodeToString([]byte(hook.GetRefID())),
|
base64.RawURLEncoding.EncodeToString([]byte(hook.GetRefID())),
|
||||||
|
|
|
@ -117,6 +117,9 @@ func LoadLogs(runOpts *options.ServerConfig, safeID webhooks.URLSafeGitID) (*Job
|
||||||
var f *os.File = nil
|
var f *os.File = nil
|
||||||
if value, ok := Actives.Load(refID); ok {
|
if value, ok := Actives.Load(refID); ok {
|
||||||
j := value.(*Job)
|
j := value.(*Job)
|
||||||
|
j.mux.Lock()
|
||||||
|
j.Logs = j.Logs[:]
|
||||||
|
j.mux.Unlock()
|
||||||
return j, nil
|
return j, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue