Add ScheduledTasks and TaskSet interfaces.

Add unlisted LogLongTick developer setting.
Rearrange the order of the shutdown signal handler to figure out where it might fail.
This commit is contained in:
Azareal 2021-05-03 10:36:29 +10:00
parent 6766fd659c
commit 77291e4b44
24 changed files with 218 additions and 89 deletions

View File

@ -21,18 +21,18 @@ func NewDefaultAgentViewCounter(acc *qgen.Accumulator) (*DefaultAgentViewCounter
buckets: make([]int64, len(agentMapEnum)),
insert: acc.Insert("viewchunks_agents").Columns("count,createdAt,browser").Fields("?,UTC_TIMESTAMP(),?").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick)
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}
func (co *DefaultAgentViewCounter) Tick() error {
for id, _ := range co.buckets {
count := atomic.SwapInt64(&co.buckets[id], 0)
err := co.insertChunk(count, id) // TODO: Bulk insert for speed?
if err != nil {
return errors.Wrap(errors.WithStack(err), "agent counter")
e := co.insertChunk(count, id) // TODO: Bulk insert for speed?
if e != nil {
return errors.Wrap(errors.WithStack(e), "agent counter")
}
}
return nil
@ -44,13 +44,13 @@ func (co *DefaultAgentViewCounter) insertChunk(count int64, agent int) error {
}
agentName := reverseAgentMapEnum[agent]
c.DebugLogf("Inserting a vchunk with a count of %d for agent %s (%d)", count, agentName, agent)
_, err := co.insert.Exec(count, agentName)
return err
_, e := co.insert.Exec(count, agentName)
return e
}
func (co *DefaultAgentViewCounter) Bump(agent int) {
// TODO: Test this check
c.DebugDetail("buckets[", agent, "]: ", co.buckets[agent])
c.DebugDetail("buckets ", agent, ": ", co.buckets[agent])
if len(co.buckets) <= agent || agent < 0 {
return
}

View File

@ -5,7 +5,7 @@ import (
"sync"
c "github.com/Azareal/Gosora/common"
"github.com/Azareal/Gosora/query_gen"
qgen "github.com/Azareal/Gosora/query_gen"
"github.com/pkg/errors"
)
@ -29,9 +29,9 @@ func NewDefaultForumViewCounter() (*DefaultForumViewCounter, error) {
evenMap: make(map[int]*RWMutexCounterBucket),
insert: acc.Insert("viewchunks_forums").Columns("count,createdAt,forum").Fields("?,UTC_TIMESTAMP(),?").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick) // There could be a lot of routes, so we don't want to be running this every second
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick) // There could be a lot of routes, so we don't want to be running this every second
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}

View File

@ -118,9 +118,9 @@ func NewDefaultLangViewCounter(acc *qgen.Accumulator) (*DefaultLangViewCounter,
insert: acc.Insert("viewchunks_langs").Columns("count,createdAt,lang").Fields("?,UTC_TIMESTAMP(),?").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick)
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}

View File

@ -30,9 +30,9 @@ func NewMemoryCounter(acc *qgen.Accumulator) (*DefaultMemoryCounter, error) {
co := &DefaultMemoryCounter{
insert: acc.Insert("memchunks").Columns("count,stack,heap,createdAt").Fields("?,?,?,UTC_TIMESTAMP()").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick)
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
ticker := time.NewTicker(time.Minute)
go func() {
for {

View File

@ -37,9 +37,9 @@ func NewDefaultPerfCounter(acc *qgen.Accumulator) (*DefaultPerfCounter, error) {
insert: acc.Insert("perfchunks").Columns("low,high,avg,createdAt").Fields("?,?,?,UTC_TIMESTAMP()").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick)
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}

View File

@ -24,9 +24,9 @@ func NewPostCounter() (*DefaultPostCounter, error) {
currentBucket: 0,
insert: acc.Insert("postchunks").Columns("count,createdAt").Fields("?,UTC_TIMESTAMP()").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick)
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}

View File

@ -37,9 +37,9 @@ func NewDefaultReferrerTracker() (*DefaultReferrerTracker, error) {
even: make(map[string]*ReferrerItem),
insert: acc.Insert("viewchunks_referrers").Columns("count,createdAt,domain").Fields("?,UTC_TIMESTAMP(),?").Prepare(), // TODO: Do something more efficient than doing a query for each referrer
}
c.AddScheduledFifteenMinuteTask(refTracker.Tick)
//c.AddScheduledSecondTask(refTracker.Tick)
c.AddShutdownTask(refTracker.Tick)
c.Tasks.FifteenMin.Add(refTracker.Tick)
//c.Tasks.Sec.Add(refTracker.Tick)
c.Tasks.Shutdown.Add(refTracker.Tick)
return refTracker, acc.FirstError()
}

View File

@ -25,9 +25,9 @@ func NewGlobalViewCounter(acc *qgen.Accumulator) (*DefaultViewCounter, error) {
currentBucket: 0,
insert: acc.Insert("viewchunks").Columns("count,createdAt,route").Fields("?,UTC_TIMESTAMP(),''").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick) // This is run once every fifteen minutes to match the frequency of the RouteViewCounter
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick) // This is run once every fifteen minutes to match the frequency of the RouteViewCounter
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}

View File

@ -41,9 +41,9 @@ func NewDefaultRouteViewCounter(acc *qgen.Accumulator) (*DefaultRouteViewCounter
insert5: acc.BulkInsert("viewchunks").Columns("count,avg,createdAt,route").Fields(fields, fields, fields, fields, fields).Prepare(),
}
if !c.Config.DisableAnalytics {
c.AddScheduledFifteenMinuteTask(co.Tick) // There could be a lot of routes, so we don't want to be running this every second
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick) // There could be a lot of routes, so we don't want to be running this every second
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
}
return co, acc.FirstError()
}

View File

@ -21,18 +21,17 @@ func NewDefaultOSViewCounter(acc *qgen.Accumulator) (*DefaultOSViewCounter, erro
buckets: make([]int64, len(osMapEnum)),
insert: acc.Insert("viewchunks_systems").Columns("count,createdAt,system").Fields("?,UTC_TIMESTAMP(),?").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick)
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}
func (co *DefaultOSViewCounter) Tick() error {
for id, _ := range co.buckets {
count := atomic.SwapInt64(&co.buckets[id], 0)
err := co.insertChunk(count, id) // TODO: Bulk insert for speed?
if err != nil {
return errors.Wrap(errors.WithStack(err), "system counter")
if e := co.insertChunk(count, id); e != nil { // TODO: Bulk insert for speed?
return errors.Wrap(errors.WithStack(e), "system counter")
}
}
return nil

View File

@ -24,9 +24,9 @@ func NewTopicCounter() (*DefaultTopicCounter, error) {
currentBucket: 0,
insert: acc.Insert("topicchunks").Columns("count,createdAt").Fields("?,UTC_TIMESTAMP()").Prepare(),
}
c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(co.Tick)
c.Tasks.FifteenMin.Add(co.Tick)
//c.Tasks.Sec.Add(co.Tick)
c.Tasks.Shutdown.Add(co.Tick)
return co, acc.FirstError()
}

View File

@ -28,6 +28,9 @@ type DefaultTopicViewCounter struct {
resetOdd *sql.Stmt
resetEven *sql.Stmt
resetBoth *sql.Stmt
insertListBuf []TopicViewInsert
saveTick *SavedTick
}
func NewDefaultTopicViewCounter() (*DefaultTopicViewCounter, error) {
@ -42,27 +45,68 @@ func NewDefaultTopicViewCounter() (*DefaultTopicViewCounter, error) {
resetOdd: acc.Update(t).Set("weekOddViews=0").Prepare(),
resetEven: acc.Update(t).Set("weekEvenViews=0").Prepare(),
resetBoth: acc.Update(t).Set("weekOddViews=0,weekEvenViews=0").Prepare(),
//insertListBuf: make([]TopicViewInsert, 1024),
}
err := co.WeekResetInit()
if err != nil {
return co, err
e := co.WeekResetInit()
if e != nil {
return co, e
}
addTick := func(f func() error) {
c.AddScheduledFifteenMinuteTask(f) // Who knows how many topics we have queued up, we probably don't want this running too frequently
//c.AddScheduledSecondTask(f)
c.AddShutdownTask(f)
tick := func(f func() error) {
c.Tasks.FifteenMin.Add(f) // Who knows how many topics we have queued up, we probably don't want this running too frequently
//c.Tasks.Sec.Add(f)
c.Tasks.Shutdown.Add(f)
}
addTick(co.Tick)
addTick(co.WeekResetTick)
tick(co.Tick)
tick(co.WeekResetTick)
return co, acc.FirstError()
}
type TopicViewInsert struct {
Count int
TopicID int
}
type SavedTick struct {
I int
I2 int
}
func (co *DefaultTopicViewCounter) handleInsertListBuf(i, i2 int) error {
ilb := co.insertListBuf
var lastSuccess int
for i3 := i2; i3 < i; i3++ {
iitem := ilb[i3]
if e := co.insertChunk(iitem.Count, iitem.TopicID); e != nil {
co.saveTick = &SavedTick{I: i, I2: lastSuccess + 1}
for i3 := i2; i3 < i && i3 <= lastSuccess; i3++ {
ilb[i3].Count, ilb[i3].TopicID = 0, 0
}
return errors.Wrap(errors.WithStack(e), "topicview counter")
}
lastSuccess = i3
}
for i3 := i2; i3 < i; i3++ {
ilb[i3].Count, ilb[i3].TopicID = 0, 0
}
return nil
}
func (co *DefaultTopicViewCounter) Tick() error {
// TODO: Fold multiple 1 view topics into one query
/*if co.saveTick != nil {
e := co.handleInsertListBuf(co.saveTick.I, co.saveTick.I2)
if e != nil {
return e
}
co.saveTick = nil
}*/
cLoop := func(l *sync.RWMutex, m map[int]*RWMutexCounterBucket) error {
//i := 0
l.RLock()
for topicID, topic := range m {
l.RUnlock()
@ -74,18 +118,23 @@ func (co *DefaultTopicViewCounter) Tick() error {
l.Lock()
delete(m, topicID)
l.Unlock()
e := co.insertChunk(count, topicID)
if e != nil {
/*if len(co.insertListBuf) >= i {
co.insertListBuf[i].Count = count
co.insertListBuf[i].TopicID = topicID
i++
} else if i < 4096 {
co.insertListBuf = append(co.insertListBuf, TopicViewInsert{count, topicID})
} else */if e := co.insertChunk(count, topicID); e != nil {
return errors.Wrap(errors.WithStack(e), "topicview counter")
}
l.RLock()
}
l.RUnlock()
return nil
return nil //co.handleInsertListBuf(i, 0)
}
err := cLoop(&co.oddLock, co.oddTopics)
if err != nil {
return err
e := cLoop(&co.oddLock, co.oddTopics)
if e != nil {
return e
}
return cLoop(&co.evenLock, co.evenTopics)
}

View File

@ -55,7 +55,7 @@ func NewDefaultGroupPromotionStore(acc *qgen.Accumulator) (*DefaultGroupPromotio
updateUser: acc.Update("users").Set("group=?").Where("group=? AND uid=?").Prepare(),
updateGeneric: acc.Update("users").Set("group=?").Where("group=? AND level>=? AND posts>=?").Prepare(),
}
AddScheduledFifteenMinuteTask(prs.Tick)
Tasks.FifteenMin.Add(prs.Tick)
return prs, acc.FirstError()
}

View File

@ -157,6 +157,7 @@ type devConfig struct {
//QuicPort int // Experimental!
//ExpFix1 bool // unlisted setting, experimental fix for http/1.1 conn hangs
LogLongTick bool // unlisted setting
}
// configHolder is purely for having a big struct to unmarshal data into

View File

@ -19,12 +19,66 @@ type TaskStmts struct {
getSync *sql.Stmt
}
var ScheduledHalfSecondTasks []func() error
var Tasks *ScheduledTasks
type TaskSet interface {
Add(func() error)
GetList() []func() error
Run() error
Count() int
}
type DefaultTaskSet struct {
Tasks []func() error
}
func (s *DefaultTaskSet) Add(task func() error) {
s.Tasks = append(s.Tasks, task)
}
func (s *DefaultTaskSet) GetList() []func() error {
return s.Tasks
}
func (s *DefaultTaskSet) Run() error {
for _, task := range s.Tasks {
if e := task(); e != nil {
return e
}
}
return nil
}
func (s *DefaultTaskSet) Count() int {
return len(s.Tasks)
}
type ScheduledTasks struct {
HalfSec TaskSet
Sec TaskSet
FifteenMin TaskSet
Hour TaskSet
Day TaskSet
Shutdown TaskSet
}
func NewScheduledTasks() *ScheduledTasks {
return &ScheduledTasks{
HalfSec: &DefaultTaskSet{},
Sec: &DefaultTaskSet{},
FifteenMin: &DefaultTaskSet{},
Hour: &DefaultTaskSet{},
Day: &DefaultTaskSet{},
Shutdown: &DefaultTaskSet{},
}
}
/*var ScheduledHalfSecondTasks []func() error
var ScheduledSecondTasks []func() error
var ScheduledFifteenMinuteTasks []func() error
var ScheduledHourTasks []func() error
var ScheduledDayTasks []func() error
var ShutdownTasks []func() error
var ShutdownTasks []func() error*/
var taskStmts TaskStmts
var lastSync time.Time
@ -41,7 +95,7 @@ func init() {
}
// AddScheduledHalfSecondTask is not concurrency safe
func AddScheduledHalfSecondTask(task func() error) {
/*func AddScheduledHalfSecondTask(task func() error) {
ScheduledHalfSecondTasks = append(ScheduledHalfSecondTasks, task)
}
@ -98,7 +152,7 @@ func ScheduledDayTaskCount() int {
// ShutdownTaskCount is not concurrency safe
func ShutdownTaskCount() int {
return len(ShutdownTasks)
}
}*/
// TODO: Use AddScheduledSecondTask
func HandleExpiredScheduledGroups() error {

View File

@ -20,7 +20,7 @@ type SingleServerThaw struct {
func NewSingleServerThaw() *SingleServerThaw {
t := &SingleServerThaw{}
if Config.ServerCount == 1 {
AddScheduledSecondTask(t.Tick)
Tasks.Sec.Add(t.Tick)
}
return t
}
@ -59,7 +59,7 @@ type DefaultThaw struct {
func NewDefaultThaw() *DefaultThaw {
t := &DefaultThaw{}
AddScheduledSecondTask(t.Tick)
Tasks.Sec.Add(t.Tick)
return t
}

View File

@ -196,7 +196,7 @@ func Dailies() (e error) {
if e != nil {
return e
}
if e = RunTasks(ScheduledDayTasks); e != nil {
if e = Tasks.Day.Run(); e != nil {
return e
}
e = ForumActionStore.DailyTick()

View File

@ -2,6 +2,7 @@ package common
import (
"database/sql"
"fmt"
"strconv"
"sync"
"time"
@ -83,8 +84,8 @@ func NewDefaultTopicList(acc *qgen.Accumulator) (*DefaultTopicList, error) {
return nil, err
}
AddScheduledHalfSecondTask(tList.Tick)
//AddScheduledSecondTask(tList.GroupCountTick) // TODO: Dynamically change the groups in the short list to be optimised every second
Tasks.HalfSec.Add(tList.Tick)
//Tasks.Sec.Add(tList.GroupCountTick) // TODO: Dynamically change the groups in the short list to be optimised every second
return tList, nil
}
@ -209,6 +210,7 @@ func (tList *DefaultTopicList) Tick() error {
tList.qcounts2 = qcounts2
tList.qLock2.Unlock()
fmt.Printf("Forums: %+v\n", Forums)
forums, err := Forums.GetAll()
if err != nil {
return err

View File

@ -412,9 +412,9 @@ func InitWidgets() (fi error) {
return fi
}
AddScheduledSecondTask(Docks.LeftSidebar.Scheduler.Tick)
AddScheduledSecondTask(Docks.RightSidebar.Scheduler.Tick)
AddScheduledSecondTask(Docks.Footer.Scheduler.Tick)
Tasks.Sec.Add(Docks.LeftSidebar.Scheduler.Tick)
Tasks.Sec.Add(Docks.RightSidebar.Scheduler.Tick)
Tasks.Sec.Add(Docks.Footer.Scheduler.Tick)
return nil
}

View File

@ -66,7 +66,7 @@ func (h *WsHubImpl) Start() {
return
}
h.lastTick = time.Now()
AddScheduledSecondTask(h.Tick)
Tasks.Sec.Add(h.Tick)
}
// This Tick is separate from the admin one, as we want to process that in parallel with this due to the blocking calls to gopsutil

View File

@ -61,14 +61,13 @@ func gloinit() (e error) {
ws := func(e error) error {
return errors.WithStack(e)
}
e = c.LoadConfig()
if e != nil {
if e = c.LoadConfig(); e != nil {
return ws(e)
}
e = c.ProcessConfig()
if e != nil {
if e = c.ProcessConfig(); e != nil {
return ws(e)
}
c.Tasks = c.NewScheduledTasks()
e = c.InitTemplates()
if e != nil {

View File

@ -416,6 +416,7 @@ func main() {
c.ErrLogWriter = f
c.ErrLogger = log.New(c.ErrLogWriter, "", log.LstdFlags)
}
c.Tasks = c.NewScheduledTasks()
err = c.InitTemplates()
if err != nil {
@ -596,9 +597,10 @@ func main() {
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
go func() {
sig := <-sigs
log.Print("Received a signal to shutdown: ", sig)
// TODO: Gracefully shutdown the HTTP server
c.RunTasks(c.ShutdownTasks)
c.StoppedServer("Received a signal to shutdown: ", sig)
c.Tasks.Shutdown.Run()
c.StoppedServer("Stopped server")
}()
// Start up the WebSocket ticks

View File

@ -171,7 +171,7 @@ func NewGenRouter(cfg *RouterConfig) (*GenRouter, error) {
suspLog: suspReqLog,
}
if !cfg.DisableTick {
c.AddScheduledDayTask(ro.DailyTick)
c.Tasks.Day.Add(ro.DailyTick)
}
return ro, nil
}

View File

@ -7,6 +7,7 @@ import (
"time"
c "github.com/Azareal/Gosora/common"
"github.com/Azareal/Gosora/uutils"
"github.com/pkg/errors"
)
@ -35,6 +36,16 @@ func deferredDailies() error {
return nil
}
func handleLogLongTick(name string, cn int64) {
if !c.Dev.LogLongTick {
return
}
dur := time.Duration(uutils.Nanotime() - cn)
if dur.Seconds() > 5 {
log.Print("tick " + name + " completed in " + dur.String())
}
}
func tickLoop(thumbChan chan bool) error {
tl := c.NewTickLoop()
TickLoop = tl
@ -45,26 +56,28 @@ func tickLoop(thumbChan chan bool) error {
return e
}
tick := func(name string, tasks []func() error) error {
tick := func(name string, tasks c.TaskSet) error {
if c.StartTick() {
return nil
}
if e := runHook("before_" + name + "_tick"); e != nil {
return e
}
if e := c.RunTasks(tasks); e != nil {
cn := uutils.Nanotime()
if e := tasks.Run(); e != nil {
return e
}
handleLogLongTick(name, cn)
return runHook("after_" + name + "_tick")
}
tl.HalfSecf = func() error {
return tick("half_second", c.ScheduledHalfSecondTasks)
return tick("half_second", c.Tasks.HalfSec)
}
// TODO: Automatically lock topics, if they're really old, and the associated setting is enabled.
// TODO: Publish scheduled posts.
tl.FifteenMinf = func() error {
return tick("fifteen_minute", c.ScheduledFifteenMinuteTasks)
return tick("fifteen_minute", c.Tasks.FifteenMin)
}
// TODO: Handle the instance going down a lot better
// TODO: Handle the daily clean-up.
@ -72,7 +85,12 @@ func tickLoop(thumbChan chan bool) error {
if c.StartTick() {
return nil
}
return c.Dailies()
cn := uutils.Nanotime()
if e := c.Dailies(); e != nil {
return e
}
handleLogLongTick("day", cn)
return nil
}
tl.Secf = func() (e error) {
@ -82,8 +100,10 @@ func tickLoop(thumbChan chan bool) error {
if e = runHook("before_second_tick"); e != nil {
return e
}
cn := uutils.Nanotime()
go func() { thumbChan <- true }()
if e = c.RunTasks(c.ScheduledSecondTasks); e != nil {
if e = c.Tasks.Sec.Run(); e != nil {
return e
}
@ -98,6 +118,7 @@ func tickLoop(thumbChan chan bool) error {
if e = c.HandleServerSync(); e != nil {
return e
}
handleLogLongTick("second", cn)
// TODO: Manage the TopicStore, UserStore, and ForumStore
// TODO: Alert the admin, if CPU usage, RAM usage, or the number of posts in the past second are too high
@ -113,6 +134,7 @@ func tickLoop(thumbChan chan bool) error {
if e := runHook("before_hour_tick"); e != nil {
return e
}
cn := uutils.Nanotime()
jsToken, e := c.GenerateSafeString(80)
if e != nil {
@ -127,9 +149,10 @@ func tickLoop(thumbChan chan bool) error {
}
c.SessionSigningKeyBox.Store(sessionSigningKey)
if e = c.RunTasks(c.ScheduledHourTasks); e != nil {
if e = c.Tasks.Hour.Run(); e != nil {
return e
}
handleLogLongTick("hour", cn)
return runHook("after_hour_tick")
}