apps/core
1
0
Fork 0
mirror of https://github.com/datarhei/core.git synced 2025-09-26 20:11:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix locking for process map

Browse Source
This commit is contained in:
Ingo Oppermann
2025-07-17 16:43:37 +02:00
parent d93cfc7564
commit 8d2f0b2c16
5 changed files with 180 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
restream/app/process.go
View File

@@ -482,6 +482,11 @@ func ParseProcessID(pid string) ProcessID {
}
func (p ProcessID) String() string {
// TODO: causes problems in the cluster with processes without a domain
//if len(p.Domain) == 0 {
// return p.ID
//}
return p.ID + "@" + p.Domain
}
@@ -498,6 +503,8 @@ func (p *ProcessID) Parse(pid string) {
if i == -1 {
p.ID = pid
p.Domain = ""
return
}
p.ID = pid[:i]

155
restream/core.go
View File

@@ -183,9 +183,7 @@ func (r *restream) Start() {
go r.resourceObserver(ctx, r.resources, time.Second)
}
r.tasks.Range(true, func(id app.ProcessID, t *task, token string) bool {
defer t.Release(token)
r.tasks.Range(func(id app.ProcessID, t *task) bool {
t.Restore()
// The filesystem cleanup rules can be set
@@ -216,10 +214,9 @@ func (r *restream) Stop() {
// Stop the currently running processes without altering their order such that on a subsequent
// Start() they will get restarted.
r.tasks.Range(true, func(_ app.ProcessID, t *task, token string) bool {
r.tasks.Range(func(_ app.ProcessID, t *task) bool {
wg.Add(1)
go func(t *task) {
defer t.Release(token)
defer wg.Done()
t.Kill()
}(t)
@@ -229,8 +226,7 @@ func (r *restream) Stop() {
wg.Wait()
r.tasks.Range(true, func(id app.ProcessID, t *task, token string) bool {
defer t.Release(token)
r.tasks.Range(func(id app.ProcessID, t *task) bool {
r.unsetCleanup(id)
return true
})
@@ -263,8 +259,7 @@ func (r *restream) filesystemObserver(ctx context.Context, fs fs.Filesystem, int
if isFull {
// Stop all tasks that write to this filesystem
r.tasks.Range(true, func(id app.ProcessID, t *task, token string) bool {
defer t.Release(token)
r.tasks.Range(func(id app.ProcessID, t *task) bool {
if !t.UsesDisk() {
return true
}
@@ -320,8 +315,7 @@ func (r *restream) resourceObserver(ctx context.Context, rsc resources.Resources
break
}
r.tasks.Range(true, func(id app.ProcessID, t *task, token string) bool {
defer t.Release(token)
r.tasks.Range(func(id app.ProcessID, t *task) bool {
limitGPU := false
gpuindex := t.GetHWDevice()
if gpuindex >= 0 && len(limitGPUs) >= gpuindex+1 {
@@ -381,7 +375,7 @@ func (r *restream) load() error {
// Replace all placeholders in the config
resolveStaticPlaceholders(t.config, r.replace)
tasks.Store(t.ID(), t)
tasks.LoadAndStore(t.ID(), t)
}
}
@@ -389,9 +383,7 @@ func (r *restream) load() error {
// replaced, we can resolve references and validate the
// inputs and outputs.
tasks.Range(false, func(_ app.ProcessID, t *task, token string) bool {
defer t.Release(token)
tasks.Range(func(_ app.ProcessID, t *task) bool {
// Just warn if the ffmpeg version constraint doesn't match the available ffmpeg version
if c, err := semver.NewConstraint(t.config.FFVersion); err == nil {
if v, err := semver.NewVersion(skills.FFmpeg.Version); err == nil {
@@ -468,12 +460,14 @@ func (r *restream) load() error {
}
t.ffmpeg = ffmpeg
t.SetValid(true)
return true
})
r.tasks.Clear()
r.tasks.Clear(func(_ app.ProcessID, t *task) bool {
t.Destroy()
return true
})
r.tasks = tasks
r.metadata = data.Metadata
@@ -488,13 +482,7 @@ func (r *restream) save() {
data := store.NewData()
r.tasks.Range(true, func(tid app.ProcessID, t *task, token string) bool {
defer t.Release(token)
if !t.IsValid() {
return true
}
r.tasks.Range(func(tid app.ProcessID, t *task) bool {
domain := data.Process[tid.Domain]
if domain == nil {
domain = map[string]store.Process{}
@@ -528,13 +516,20 @@ func (r *restream) CreatedAt() time.Time {
}
func (r *restream) AddProcess(config *app.Config) error {
tid := app.ProcessID{
ID: config.ID,
Domain: config.Domain,
}
if r.tasks.Has(tid) {
return ErrProcessExists
}
t, err := r.createTask(config)
if err != nil {
return err
}
tid := t.ID()
_, ok := r.tasks.LoadOrStore(tid, t)
if ok {
t.Destroy()
@@ -546,7 +541,7 @@ func (r *restream) AddProcess(config *app.Config) error {
err = t.Restore()
if err != nil {
r.tasks.Delete(tid)
r.tasks.LoadAndDelete(tid)
t.Destroy()
return err
}
@@ -655,8 +650,6 @@ func (r *restream) createTask(config *app.Config) (*task, error) {
t.ffmpeg = ffmpeg
t.SetValid(true)
return t, nil
}
@@ -683,9 +676,8 @@ func (r *restream) onBeforeStart(cfg *app.Config) func([]string) ([]string, erro
selectedGPU = 0
}
if t, token, hasTask := r.tasks.Load(cfg.ProcessID()); hasTask {
if t, hasTask := r.tasks.LoadUnsafe(cfg.ProcessID()); hasTask {
t.SetHWDevice(selectedGPU)
t.Release(token)
} else {
return []string{}, fmt.Errorf("process with the ID '%s' not found", cfg.ProcessID())
}
@@ -1046,16 +1038,12 @@ func (r *restream) resolveAddress(tasks *Storage, id, address string) (string, e
}
var t *task = nil
var ttoken string = ""
tasks.Range(true, func(_ app.ProcessID, task *task, token string) bool {
tasks.Range(func(_ app.ProcessID, task *task) bool {
if task.id == matches["id"] && task.domain == matches["domain"] {
t = task
ttoken = token
return false
}
task.Release(token)
return true
})
@@ -1063,8 +1051,6 @@ func (r *restream) resolveAddress(tasks *Storage, id, address string) (string, e
return address, fmt.Errorf("unknown process '%s' in domain '%s' (%s): %w", matches["id"], matches["domain"], address, ErrInvalidProcessConfig)
}
defer t.Release(ttoken)
teeOptions := regexp.MustCompile(`^\[[^\]]*\]`)
for _, x := range t.config.Output {
@@ -1176,9 +1162,9 @@ func (r *restream) UpdateProcess(id app.ProcessID, config *app.Config) error {
return ErrUnknownProcess
}
err := r.updateProcess(task, config)
defer r.tasks.Unlock(id)
task.Unlock()
err := r.updateProcess(task, config)
if err != nil {
return err
@@ -1217,7 +1203,7 @@ func (r *restream) updateProcess(task *task, config *app.Config) error {
t.process.Order.Set(order)
if err := r.stopProcess(task); err != nil {
if err := task.Stop(); err != nil {
t.Destroy()
return fmt.Errorf("stop process: %w", err)
}
@@ -1231,21 +1217,22 @@ func (r *restream) updateProcess(task *task, config *app.Config) error {
// Transfer the metadata to the new process
t.ImportMetadata(task.ExportMetadata())
if err := r.deleteProcess(task); err != nil {
t.Destroy()
return fmt.Errorf("delete process: %w", err)
}
r.unsetPlayoutPorts(task)
r.unsetCleanup(task.ID())
t.Lock()
defer t.Unlock()
r.tasks.Store(tid, t)
r.tasks.LoadAndStore(tid, t)
// set filesystem cleanup rules
r.setCleanup(tid, t.config)
t.Restore()
if !tid.Equal(task.ID()) {
r.tasks.LoadAndDelete(task.ID())
}
task.Destroy()
return nil
}
@@ -1276,9 +1263,7 @@ func (r *restream) GetProcessIDs(idpattern, refpattern, ownerpattern, domainpatt
if idglob == nil && refglob == nil && ownerglob == nil && domainglob == nil {
ids = make([]app.ProcessID, 0, r.tasks.Size())
r.tasks.Range(true, func(id app.ProcessID, t *task, token string) bool {
defer t.Release(token)
r.tasks.Range(func(id app.ProcessID, t *task) bool {
ids = append(ids, id)
return true
@@ -1286,9 +1271,7 @@ func (r *restream) GetProcessIDs(idpattern, refpattern, ownerpattern, domainpatt
} else {
ids = []app.ProcessID{}
r.tasks.Range(true, func(id app.ProcessID, t *task, token string) bool {
defer t.Release(token)
r.tasks.Range(func(id app.ProcessID, t *task) bool {
if !t.Match(idglob, refglob, ownerglob, domainglob) {
return true
}
@@ -1303,11 +1286,10 @@ func (r *restream) GetProcessIDs(idpattern, refpattern, ownerpattern, domainpatt
}
func (r *restream) GetProcess(id app.ProcessID) (*app.Process, error) {
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadUnsafe(id)
if !ok {
return &app.Process{}, ErrUnknownProcess
}
defer task.Release(token)
return task.Process(), nil
}
@@ -1320,7 +1302,7 @@ func (r *restream) DeleteProcess(id app.ProcessID) error {
err := r.deleteProcess(task)
task.Unlock()
r.tasks.Unlock(id)
if err != nil {
return err
@@ -1339,7 +1321,7 @@ func (r *restream) deleteProcess(task *task) error {
r.unsetPlayoutPorts(task)
r.unsetCleanup(task.ID())
r.tasks.Delete(task.ID())
r.tasks.LoadAndDelete(task.ID())
task.Destroy()
@@ -1347,15 +1329,14 @@ func (r *restream) deleteProcess(task *task) error {
}
func (r *restream) StartProcess(id app.ProcessID) error {
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadAndLock(id)
if !ok {
return ErrUnknownProcess
}
defer r.tasks.Unlock(id)
err := r.startProcess(task)
task.Release(token)
if err != nil {
return err
}
@@ -1377,15 +1358,14 @@ func (r *restream) startProcess(task *task) error {
}
func (r *restream) StopProcess(id app.ProcessID) error {
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadAndLock(id)
if !ok {
return ErrUnknownProcess
}
defer r.tasks.Unlock(id)
err := r.stopProcess(task)
task.Release(token)
if err != nil {
return err
}
@@ -1408,11 +1388,11 @@ func (r *restream) stopProcess(task *task) error {
}
func (r *restream) RestartProcess(id app.ProcessID) error {
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadAndLock(id)
if !ok {
return ErrUnknownProcess
}
defer task.Release(token)
defer r.tasks.Unlock(id)
return r.restartProcess(task)
}
@@ -1431,7 +1411,7 @@ func (r *restream) ReloadProcess(id app.ProcessID) error {
err := r.reloadProcess(task)
task.Unlock()
r.tasks.Unlock(id)
if err != nil {
return err
@@ -1464,34 +1444,33 @@ func (r *restream) reloadProcess(task *task) error {
}
// Transfer the report history to the new process
t.parser.ImportReportHistory(task.parser.ReportHistory())
t.ImportParserReportHistory(task.ExportParserReportHistory())
// Transfer the metadata to the new process
t.metadata = task.metadata
t.ImportMetadata(task.ExportMetadata())
if err := r.deleteProcess(task); err != nil {
t.Destroy()
return fmt.Errorf("delete process: %w", err)
}
r.unsetPlayoutPorts(task)
r.unsetCleanup(task.ID())
r.tasks.Store(tid, t)
r.tasks.LoadAndStore(tid, t)
// set filesystem cleanup rules
r.setCleanup(tid, t.config)
t.Restore()
task.Destroy()
return nil
}
func (r *restream) GetProcessState(id app.ProcessID) (*app.State, error) {
state := &app.State{}
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadUnsafe(id)
if !ok {
return state, ErrUnknownProcess
}
defer task.Release(token)
return task.State()
}
@@ -1499,11 +1478,10 @@ func (r *restream) GetProcessState(id app.ProcessID) (*app.State, error) {
func (r *restream) GetProcessReport(id app.ProcessID) (*app.Report, error) {
report := &app.Report{}
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadUnsafe(id)
if !ok {
return report, ErrUnknownProcess
}
defer task.Release(token)
return task.Report()
}
@@ -1513,7 +1491,7 @@ func (r *restream) SetProcessReport(id app.ProcessID, report *app.Report) error
if !ok {
return ErrUnknownProcess
}
defer task.Unlock()
defer r.tasks.Unlock(id)
return task.SetReport(report)
}
@@ -1524,15 +1502,13 @@ func (r *restream) SearchProcessLogHistory(idpattern, refpattern, state string,
ids := r.GetProcessIDs(idpattern, refpattern, "", "")
for _, id := range ids {
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadUnsafe(id)
if !ok {
continue
}
presult := task.SearchReportHistory(state, from, to)
result = append(result, presult...)
task.Release(token)
}
return result
@@ -1626,15 +1602,11 @@ func (r *restream) ReloadSkills() error {
}
func (r *restream) GetPlayout(id app.ProcessID, inputid string) (string, error) {
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadAndRLock(id)
if !ok {
return "", ErrUnknownProcess
}
defer task.Release(token)
if !task.IsValid() {
return "", ErrInvalidProcessConfig
}
defer r.tasks.RUnlock(id)
port, ok := task.playout[inputid]
if !ok {
@@ -1652,7 +1624,7 @@ func (r *restream) SetProcessMetadata(id app.ProcessID, key string, data interfa
err := task.SetMetadata(key, data)
task.Unlock()
r.tasks.Unlock(id)
if err != nil {
return err
@@ -1664,11 +1636,10 @@ func (r *restream) SetProcessMetadata(id app.ProcessID, key string, data interfa
}
func (r *restream) GetProcessMetadata(id app.ProcessID, key string) (interface{}, error) {
task, token, ok := r.tasks.Load(id)
task, ok := r.tasks.LoadUnsafe(id)
if !ok {
return nil, ErrUnknownProcess
}
defer task.Release(token)
return task.GetMetadata(key)
}

14
restream/core_test.go
View File

@@ -988,15 +988,15 @@ func TestTeeAddressReference(t *testing.T) {
r := rs.(*restream)
task, _, ok := r.tasks.Load(app.ProcessID{ID: "process2"})
task, ok := r.tasks.LoadAndRLock(app.ProcessID{ID: "process2"})
require.True(t, ok)
require.Equal(t, "http://example.com/live.m3u8", task.config.Input[0].Address)
task, _, ok = r.tasks.Load(app.ProcessID{ID: "process3"})
task, ok = r.tasks.LoadAndRLock(app.ProcessID{ID: "process3"})
require.True(t, ok)
require.Equal(t, "http://example.com/live.m3u8", task.config.Input[0].Address)
task, _, ok = r.tasks.Load(app.ProcessID{ID: "process4"})
task, ok = r.tasks.LoadAndRLock(app.ProcessID{ID: "process4"})
require.True(t, ok)
require.Equal(t, "rtmp://example.com/live.stream?token=123", task.config.Input[0].Address)
}
@@ -1598,9 +1598,8 @@ func TestProcessReplacer(t *testing.T) {
LogPatterns: []string{},
}
task, token, ok := rs.tasks.Load(app.ProcessID{ID: "314159265359"})
task, ok := rs.tasks.LoadAndRLock(app.ProcessID{ID: "314159265359"})
require.True(t, ok)
task.Release(token)
require.Equal(t, process, task.config)
@@ -1693,9 +1692,8 @@ func TestProcessLimit(t *testing.T) {
rs := rsi.(*restream)
task, token, ok := rs.tasks.Load(app.ProcessID{ID: process.ID})
task, ok := rs.tasks.LoadAndRLock(app.ProcessID{ID: process.ID})
require.True(t, ok)
task.Release(token)
status := task.ffmpeg.Status()
@@ -1850,7 +1848,7 @@ func TestProcessCleanup(t *testing.T) {
rsi.Stop()
for i := 0; i < 10; i++ {
for i := range 10 {
memfs.WriteFileReader(fmt.Sprintf("/foobar_%02d.dat", i), bytes.NewReader([]byte("hello")), -1)
}

144
restream/manager.go
View File

@@ -1,97 +1,161 @@
package restream
import (
"sync"
"github.com/datarhei/core/v16/restream/app"
"github.com/puzpuzpuz/xsync/v3"
)
type metatask struct {
task *task
lock sync.RWMutex
}
type Storage struct {
tasks *xsync.MapOf[app.ProcessID, *task]
lock *xsync.RBMutex
tasks *xsync.MapOf[app.ProcessID, *metatask]
}
func NewStorage() *Storage {
m := &Storage{
tasks: xsync.NewMapOf[app.ProcessID, *task](),
lock: xsync.NewRBMutex(),
tasks: xsync.NewMapOf[app.ProcessID, *metatask](),
}
return m
}
func (m *Storage) Range(onlyValid bool, f func(key app.ProcessID, value *task, token string) bool) {
m.tasks.Range(func(id app.ProcessID, task *task) bool {
token := task.RLock()
if onlyValid && !task.IsValid() {
task.Release(token)
return true
}
return f(id, task, token)
// Range calls f sequentially for each key and value present in the map. If f returns false, range stops the iteration.
//
// Range does not necessarily correspond to any consistent snapshot of the Map's contents: no key will be visited more than once,
// but if the value for any key is stored or deleted concurrently, Range may reflect any mapping for that key from any point during the Range call.
//
// It is safe to modify the map while iterating it, including entry creation, modification and deletion. However,
// the concurrent modification rule apply, i.e. the changes may be not reflected in the subsequently iterated entries.
func (m *Storage) Range(f func(key app.ProcessID, value *task) bool) {
m.tasks.Range(func(id app.ProcessID, mt *metatask) bool {
return f(id, mt.task)
})
}
func (m *Storage) Store(id app.ProcessID, t *task) {
t, ok := m.tasks.LoadAndStore(id, t)
// Store sets the value for a key.
func (m *Storage) LoadAndStore(id app.ProcessID, t *task) (*task, bool) {
m.lock.Lock()
defer m.lock.Unlock()
mt, ok := m.tasks.Load(id)
if ok {
t.Destroy()
old := mt.task
mt.task = t
return old, true
}
mt = &metatask{
task: t,
}
m.tasks.Store(id, mt)
return t, false
}
// LoadOrStore returns the existing value for the key if present. Otherwise, it stores and returns the given value.
// The loaded result is true if the value was loaded, false if stored.
func (m *Storage) LoadOrStore(id app.ProcessID, t *task) (*task, bool) {
return m.tasks.LoadOrStore(id, t)
m.lock.Lock()
defer m.lock.Unlock()
mt, ok := m.tasks.Load(id)
if ok {
return mt.task, true
}
mt = &metatask{
task: t,
}
m.tasks.Store(id, mt)
return t, false
}
// Has returns whether a value is stored in the map.
func (m *Storage) Has(id app.ProcessID) bool {
_, hasTask := m.tasks.Load(id)
return hasTask
}
func (m *Storage) Load(id app.ProcessID) (*task, string, bool) {
task, ok := m.tasks.Load(id)
func (m *Storage) LoadUnsafe(id app.ProcessID) (*task, bool) {
mt, ok := m.tasks.Load(id)
if !ok {
return nil, "", false
return nil, false
}
token := task.RLock()
if !task.IsValid() {
task.Release(token)
return nil, "", false
return mt.task, true
}
// LoadAndRLock returns the value stored in the map for a key, or zero value of type V if no value is present.
// The ok result indicates whether value was found in the map.
func (m *Storage) LoadAndRLock(id app.ProcessID) (*task, bool) {
mt, ok := m.tasks.Load(id)
if !ok {
return nil, false
}
return task, token, true
mt.lock.RLock()
return mt.task, true
}
func (m *Storage) RUnlock(id app.ProcessID) {
mt, ok := m.tasks.Load(id)
if !ok {
return
}
mt.lock.RUnlock()
}
func (m *Storage) LoadAndLock(id app.ProcessID) (*task, bool) {
task, ok := m.tasks.Load(id)
mt, ok := m.tasks.Load(id)
if !ok {
return nil, false
}
task.Lock()
if !task.IsValid() {
task.Unlock()
return nil, false
}
return task, true
mt.lock.Lock()
return mt.task, true
}
func (m *Storage) Delete(id app.ProcessID) bool {
if t, ok := m.tasks.Load(id); ok {
m.tasks.Delete(id)
t.Destroy()
return true
func (m *Storage) Unlock(id app.ProcessID) {
mt, ok := m.tasks.Load(id)
if !ok {
return
}
return false
mt.lock.Unlock()
}
// Delete deletes the value for a key.
func (m *Storage) LoadAndDelete(id app.ProcessID) (*task, bool) {
m.lock.Lock()
defer m.lock.Unlock()
if mt, ok := m.tasks.LoadAndDelete(id); ok {
return mt.task, true
}
return nil, false
}
// Size returns current size of the map.
func (m *Storage) Size() int {
return m.tasks.Size()
}
func (m *Storage) Clear() {
m.tasks.Range(func(_ app.ProcessID, t *task) bool {
t.Destroy()
return true
// Clear deletes all keys and values currently stored in the map.
func (m *Storage) Clear(f func(key app.ProcessID, value *task) bool) {
m.tasks.Range(func(id app.ProcessID, mt *metatask) bool {
return f(id, mt.task)
})
m.tasks.Clear()

120
restream/task.go
View File

@@ -2,7 +2,6 @@ package restream
import (
"maps"
"sync"
"sync/atomic"
"time"
@@ -14,7 +13,6 @@ import (
)
type task struct {
valid *atomic.Bool // Whether the task is valid an can be used
readers *atomic.Int64 // Number of concurrent readers
id string // ID of the task/process
owner string // Owner of the process
@@ -30,13 +28,10 @@ type task struct {
usesDisk bool // Whether this task uses the disk
hwdevice *atomic.Int32 // Index of the GPU this task uses
metadata map[string]interface{} // Metadata of the process
lock sync.RWMutex
}
func NewTask(process *app.Process, logger log.Logger) *task {
t := &task{
valid: &atomic.Bool{},
readers: &atomic.Int64{},
id: process.ID,
owner: process.Owner,
@@ -53,31 +48,6 @@ func NewTask(process *app.Process, logger log.Logger) *task {
return t
}
func (t *task) Lock() {
t.lock.Lock()
}
func (t *task) Unlock() {
t.lock.Unlock()
}
func (t *task) RLock() string {
t.readers.Add(1)
return ""
}
func (t *task) Release(token string) {
t.readers.Add(-1)
}
func (t *task) IsValid() bool {
return t.valid.Load()
}
func (t *task) SetValid(valid bool) {
t.valid.Store(valid)
}
func (t *task) UsesDisk() bool {
return t.usesDisk
}
@@ -95,10 +65,6 @@ func (t *task) String() string {
// Restore restores the task's order
func (t *task) Restore() error {
if !t.valid.Load() {
return ErrInvalidProcessConfig
}
if t.process.Order.String() == "start" {
err := t.ffmpeg.Start()
if err != nil {
@@ -110,16 +76,6 @@ func (t *task) Restore() error {
}
func (t *task) Start() error {
if !t.valid.Load() {
return ErrInvalidProcessConfig
}
status := t.ffmpeg.Status()
if t.process.Order.String() == "start" && status.Order == "start" {
return nil
}
t.process.Order.Set("start")
t.ffmpeg.Start()
@@ -128,16 +84,6 @@ func (t *task) Start() error {
}
func (t *task) Stop() error {
if !t.valid.Load() {
return ErrInvalidProcessConfig
}
status := t.ffmpeg.Status()
if t.process.Order.String() == "stop" && status.Order == "stop" {
return nil
}
t.process.Order.Set("stop")
t.ffmpeg.Stop(true)
@@ -147,18 +93,10 @@ func (t *task) Stop() error {
// Kill stops a process without changing the tasks order
func (t *task) Kill() {
if !t.valid.Load() {
return
}
t.ffmpeg.Stop(true)
}
func (t *task) Restart() error {
if !t.valid.Load() {
return ErrInvalidProcessConfig
}
if t.process.Order.String() == "stop" {
return nil
}
@@ -174,10 +112,6 @@ func (t *task) Restart() error {
func (t *task) State() (*app.State, error) {
state := &app.State{}
if !t.valid.Load() {
return state, nil
}
status := t.ffmpeg.Status()
state.Order = t.process.Order.String()
@@ -267,10 +201,6 @@ func assignConfigID(progress []app.ProgressIO, config []app.ConfigIO) []app.Prog
func (t *task) Report() (*app.Report, error) {
report := &app.Report{}
if !t.valid.Load() {
return report, nil
}
current := t.parser.Report()
report.UnmarshalParser(&current)
@@ -291,10 +221,6 @@ func (t *task) Report() (*app.Report, error) {
}
func (t *task) SetReport(report *app.Report) error {
if !t.valid.Load() {
return nil
}
_, history := report.MarshalParser()
t.parser.ImportReportHistory(history)
@@ -305,10 +231,6 @@ func (t *task) SetReport(report *app.Report) error {
func (t *task) SearchReportHistory(state string, from, to *time.Time) []app.ReportHistorySearchResult {
result := []app.ReportHistorySearchResult{}
if !t.valid.Load() {
return result
}
presult := t.parser.SearchReportHistory(state, from, to)
for _, f := range presult {
@@ -377,10 +299,6 @@ func (t *task) ExportMetadata() map[string]interface{} {
}
func (t *task) Limit(cpu, memory, gpu bool) bool {
if !t.valid.Load() {
return false
}
t.ffmpeg.Limit(cpu, memory, gpu)
return true
@@ -395,41 +313,19 @@ func (t *task) GetHWDevice() int {
}
func (t *task) Equal(config *app.Config) bool {
if !t.valid.Load() {
return false
}
return t.process.Config.Equal(config)
}
func (t *task) ResolvedConfig() *app.Config {
if !t.valid.Load() {
return nil
}
return t.config.Clone()
}
func (t *task) Config() *app.Config {
if !t.valid.Load() {
return nil
}
return t.process.Config.Clone()
}
func (t *task) Destroy() {
t.Stop()
t.valid.Store(false)
/*
t.process = nil
t.config = nil
t.command = nil
t.ffmpeg = nil
t.parser = nil
t.metadata = map[string]interface{}{}
*/
}
func (t *task) Match(id, reference, owner, domain glob.Glob) bool {
@@ -468,33 +364,17 @@ func (t *task) Match(id, reference, owner, domain glob.Glob) bool {
}
func (t *task) Process() *app.Process {
if !t.valid.Load() {
return nil
}
return t.process.Clone()
}
func (t *task) Order() string {
if !t.valid.Load() {
return ""
}
return t.process.Order.String()
}
func (t *task) ExportParserReportHistory() []parse.ReportHistoryEntry {
if !t.valid.Load() {
return nil
}
return t.parser.ReportHistory()
}
func (t *task) ImportParserReportHistory(report []parse.ReportHistoryEntry) {
if !t.valid.Load() {
return
}
t.parser.ImportReportHistory(report)
}