simpletracker

README

OS Process Tacker

Introduction

OS Process Tracker implements the JobTracker interface used by the Go DRMAA2 implementation in order to use standard OS processes as a backend for managing jobs as processes from the DRMAA2 interface.

Functionality

Basic Usage

A JobTemplate requires at least:

• RemoteCommand -> which is path to the executable

Job Control Mapping

DRMAA2 Job Control	OS Process
Suspend	SIGTSTP
Resume	SIGCONT
Terminate	SIGKILL
Hold	Unsupported
Release	Unsupported

State Mapping

DRMAA2 State	Process State
Queued	Unsupported
Running	PID is found
Suspended
Done
Failed

DeleteJob

Job Template Mapping

A JobTemplate is mapped into the process creation process in the following way:

DRMAA2 JobTemplate	OS Process
RemoteCommand	Executable to start
JobName
Args	Arguments of the executable
WorkingDir	Working directory
JobEnvironment	Environment variables set

Documentation

Index

• func CreateJobInfo() drmaa2interface.JobInfo
• func GetNextJobID() string
• func KillPid(pid int) error
• func NewJobID() *lastJobID
• func ResumePid(pid int) error
• func SetJobID(jobid int64)
• func StartProcess(jobid string, t drmaa2interface.JobTemplate, finishedJobChannel chan JobEvent) (int, error)
• func SuspendPid(pid int) error
• func TrackProcess(cmd *exec.Cmd, jobid string, finishedJobChannel chan JobEvent, ...)
• type InternalJob
• type JobEvent
• type JobStore
  • func NewJobStore() *JobStore
  • func (js *JobStore) GetPID(jobid string) (int, error)
  • func (js *JobStore) SaveArrayJob(arrayjobid string, pids []int, t drmaa2interface.JobTemplate, begin int, ...)
  • func (js *JobStore) SaveJob(jobid string, t drmaa2interface.JobTemplate, pid int)
• type JobTracker
  • func New(jobsession string) *JobTracker
  • func (jt *JobTracker) AddArrayJob(t drmaa2interface.JobTemplate, begin int, end int, step int, maxParallel int) (string, error)
  • func (jt *JobTracker) AddJob(t drmaa2interface.JobTemplate) (string, error)
  • func (jt *JobTracker) DeleteJob(jobid string) error
  • func (jt *JobTracker) Destroy() error
  • func (jt *JobTracker) JobControl(jobid, state string) error
  • func (jt *JobTracker) JobInfo(jobid string) (drmaa2interface.JobInfo, error)
  • func (jt *JobTracker) JobState(jobid string) drmaa2interface.JobState
  • func (jt *JobTracker) ListArrayJobs(id string) ([]string, error)
  • func (jt *JobTracker) ListJobCategories() ([]string, error)
  • func (jt *JobTracker) ListJobs() ([]string, error)
  • func (jt *JobTracker) ProcessToJobInfo(jobid string, pid int) (drmaa2interface.JobInfo, error)
  • func (jt *JobTracker) Wait(jobid string, d time.Duration, state ...drmaa2interface.JobState) error
• type PubSub
  • func NewPubSub() (*PubSub, chan JobEvent)
  • func (ps *PubSub) Register(jobid string, states ...drmaa2interface.JobState) (chan drmaa2interface.JobState, error)
  • func (ps *PubSub) StartBookKeeper()
  • func (ps *PubSub) Unregister(id string)

Functions

func CreateJobInfo

func CreateJobInfo() drmaa2interface.JobInfo

func GetNextJobID

func GetNextJobID() string

func KillPid

func KillPid(pid int) error

func NewJobID

func NewJobID() *lastJobID

func ResumePid

func ResumePid(pid int) error

func SetJobID

func SetJobID(jobid int64)

func StartProcess

func StartProcess(jobid string, t drmaa2interface.JobTemplate, finishedJobChannel chan JobEvent) (int, error)

func SuspendPid

func SuspendPid(pid int) error

func TrackProcess

func TrackProcess(cmd *exec.Cmd, jobid string, finishedJobChannel chan JobEvent, waitForFiles int, waitCh chan bool)

Types

type InternalJob

type InternalJob struct {
	TaskID int
	State  drmaa2interface.JobState
	PID    int
}

type JobEvent

type JobEvent struct {
	JobID    string
	JobState drmaa2interface.JobState
	JobInfo  drmaa2interface.JobInfo
}

type JobStore

type JobStore struct {
	// contains filtered or unexported fields
}

func NewJobStore

func NewJobStore() *JobStore

func (*JobStore) GetPID

func (js *JobStore) GetPID(jobid string) (int, error)

func (*JobStore) SaveArrayJob

func (js *JobStore) SaveArrayJob(arrayjobid string, pids []int, t drmaa2interface.JobTemplate, begin int, end int, step int)

func (*JobStore) SaveJob

func (js *JobStore) SaveJob(jobid string, t drmaa2interface.JobTemplate, pid int)

type JobTracker

type JobTracker struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func New

func New(jobsession string) *JobTracker

func (*JobTracker) AddArrayJob

func (jt *JobTracker) AddArrayJob(t drmaa2interface.JobTemplate, begin int, end int, step int, maxParallel int) (string, error)

func (*JobTracker) AddJob

func (jt *JobTracker) AddJob(t drmaa2interface.JobTemplate) (string, error)

func (*JobTracker) DeleteJob

func (jt *JobTracker) DeleteJob(jobid string) error

TODO TEST IMPLEMENTATION

func (*JobTracker) Destroy

func (jt *JobTracker) Destroy() error

func (*JobTracker) JobControl

func (jt *JobTracker) JobControl(jobid, state string) error

func (*JobTracker) JobInfo

func (jt *JobTracker) JobInfo(jobid string) (drmaa2interface.JobInfo, error)

func (*JobTracker) JobState

func (jt *JobTracker) JobState(jobid string) drmaa2interface.JobState

func (*JobTracker) ListArrayJobs

func (jt *JobTracker) ListArrayJobs(id string) ([]string, error)

func (*JobTracker) ListJobCategories

func (jt *JobTracker) ListJobCategories() ([]string, error)

func (*JobTracker) ListJobs

func (jt *JobTracker) ListJobs() ([]string, error)

func (*JobTracker) ProcessToJobInfo

func (jt *JobTracker) ProcessToJobInfo(jobid string, pid int) (drmaa2interface.JobInfo, error)

func (*JobTracker) Wait

func (jt *JobTracker) Wait(jobid string, d time.Duration, state ...drmaa2interface.JobState) error

type PubSub

type PubSub struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewPubSub

func NewPubSub() (*PubSub, chan JobEvent)

func (*PubSub) Register

func (ps *PubSub) Register(jobid string, states ...drmaa2interface.JobState) (chan drmaa2interface.JobState, error)

func (*PubSub) StartBookKeeper

func (ps *PubSub) StartBookKeeper()

BookKeeper processes all job state changes from the process trackers and notifies registered wait functions

func (*PubSub) Unregister

func (ps *PubSub) Unregister(id string)