You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
103 lines
2.4 KiB
103 lines
2.4 KiB
package main
|
|
|
|
// thread.go: handling of worker threads
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
func init() {
|
|
registerParam("threads", 3, "how many threads to use")
|
|
registerParam("thread-delay-ms", 10, "separate threads at startup for this amount of ms")
|
|
|
|
// using a very high limit for now, but this should actually be set to -1
|
|
registerParam("task-max-deferrals", 30000, "how many deferrals are allowed for a single task. -1 to disable")
|
|
|
|
registerAlias("t", "threads")
|
|
}
|
|
|
|
// maxSafeThreads is a safeguard to prevent creation of too many threads at once.
|
|
const maxSafeThreads = 5000
|
|
|
|
// ThreadService creates, starts up and waits for all threads.
|
|
func ThreadService() {
|
|
numThreads := getParamInt("threads")
|
|
failIf(numThreads > maxSafeThreads, "too many threads (max %v)", maxSafeThreads)
|
|
|
|
log("thread", 1, "initializing %v threads", numThreads)
|
|
|
|
c := make(chan bool)
|
|
var wg sync.WaitGroup
|
|
for i := 1; i <= numThreads; i++ {
|
|
wg.Add(1)
|
|
go threadEntryPoint(c, i, &wg)
|
|
}
|
|
|
|
threadDelay := getParamDurationMS("thread-delay-ms")
|
|
log("thread", 0, "starting %v threads", numThreads)
|
|
for i := 1; i <= numThreads; i++ {
|
|
c <- true
|
|
if threadDelay > 0 {
|
|
time.Sleep(threadDelay)
|
|
}
|
|
}
|
|
|
|
wg.Wait()
|
|
log("thread", 1, "finished threads")
|
|
}
|
|
|
|
// threadEntryPoint is the main entrypoint for a work thread.
|
|
func threadEntryPoint(c chan bool, threadIdx int, wg *sync.WaitGroup) {
|
|
<-c
|
|
|
|
log("thread", 3, "starting loop for thread %v", threadIdx)
|
|
|
|
for threadWork(threadIdx) {
|
|
}
|
|
|
|
log("thread", 3, "exiting thread %v", threadIdx)
|
|
wg.Done()
|
|
}
|
|
|
|
// threadWork processes a single work item for a thread.
|
|
func threadWork(threadIdx int) bool {
|
|
task, delay := CreateTask(threadIdx)
|
|
if task == nil {
|
|
if delay > 0 {
|
|
log("thread", 3, "no work currently available, sleeping for %v in thread %v", delay, threadIdx)
|
|
time.Sleep(delay)
|
|
return true
|
|
} else if delay == 0 {
|
|
log("thread", 2, "no more work for thread %v", threadIdx)
|
|
return false
|
|
} else {
|
|
panic("negative task creation delay: " + delay.String())
|
|
}
|
|
}
|
|
|
|
log("thread", 4, "got task %v", task)
|
|
|
|
conn, err := NewConnection(task.e)
|
|
if err != nil {
|
|
task.EventWithParm(TE_NoResponse, err)
|
|
return true
|
|
}
|
|
|
|
task.Event(TN_Connected)
|
|
|
|
log("thread", 2, "trying %v", task)
|
|
|
|
res, err := TryLogin(task, conn)
|
|
if err != nil {
|
|
task.Event(TE_ProtocolError)
|
|
} else {
|
|
if res && err == nil {
|
|
task.EventWithParm(TE_Good, task.login)
|
|
} else {
|
|
task.Event(TE_Bad)
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|