Job Management

Interactive jobs, also known as synchronous jobs, can be submitted through:

• DolphinDB GUI
• DolphinDB Console (CLI)
• DolphinDB Terminal (CLI for remote server connection)
• DolphinDB Extension for Visual Studio Code
• APIs for various programming languages
• Web-based interface

Each server connection can only run one interactive job at a time, which means new jobs cannot be submitted using this connection until the last job is finished Therefore, the maximum number of concurrent interactive jobs that can be submitted to a node is determined by the number of connections the node is allowed to have (set by the configuration parameter maxConnections).

Meanwhile, the maximum number of concurrent interactive jobs that can be executed on a node is determined by the number of worker threads ("workers", set by workerNum) and web worker threads ("web workers", set by webWorkerNum) on the node. The submitted interactive jobs are added to a queue, from which the workers consume the jobs submitted via non-web clients and the web workers consume the ones submitted via the Web interface.

Interactive jobs are submitted to the node connected to the client. To submit an interactive job to another node in the cluster, call the rpc function. For example, the following script retrieves the descriptive information on the most recently finished three SQL queries on DFS databases from the remote node "node2".

rpc("node2",getCompletedQueries{3})

Call getConsoleJobs to view the running interactive jobs on the local node. For example:

>getConsoleJobs();
node	userID	rootJobId       jobType	desc	priority    parallelism receiveTime                 sessionId
------  -----   --------------  ------  ------- -------     ----------- ----------------------      -------------
NODE_1	admin	b4ff7490-9a...	unknown	unknown	0	    1           2020.03.07T20:42:04.903     323,818,682
NODE_1	guest	12c0be95-e1...	unknown	unknown	4	    2           2020.03.07T20:42:06.319     1,076,727,822

The returned table contains the following columns:

• "node", the alias of the node to which a job is submitted
• "userID", the job creator
• "priority", the job priority
• "parallelism", the job parallelism
• "receiveTime", the time when the node receives the job.

Note:

• As each connection can only have one running interactive job at the same time, we must start a new session (e.g., a new GUI window) to run getConsoleJobs on the same node.
• To check the running interactive jobs on another node, use rpc. To check the running interactive jobs on all data and compute nodes, use pnodeRun:

  pnodeRun(getConsoleJobs)

Alternatively, you can check the jobs on the DolphinDB Web interface. Starting from the 1.30.6 version, the Web interface provides the "Job" tab for browsing and canceling jobs. This tab is also available in cluster deployment. In the screenshot below, the "Job" tab has 3 tables that display the running interactive jobs, submitted batch jobs, and scheduled jobs in the cluster. Click the +/- icon to expand/collapse the details of an interactive job and click "Stop" to cancel the job.

Call cancelConsoleJob to cancel submitted but unfinished interactive job(s) by job ID.

In DolphinDB, each interactive job has a unique ID, rootJobId. If an interactive job is broken down into several tasks, the tasks take the same rootJobId as the job. To get the rootJobId of a job, call getConsoleJobs or check the "Job" tab in the Web interface as described in the previous section.

Similar to getConsoleJobs, we must create a new session to execute cancelConsoleJob. To cancel interactive jobs on other nodes in the cluster, use rpc or pnodeRun. For example, the following script defines a function that cancels all interactive jobs on the current node and calls pnodeRun to submit it to all data nodes and compute nodes.

def cancelAllConsoleJob(){
    cancelConsoleJob(getConsoleJobs()[`rootJobid])
}
pnodeRun(cancelAllConsoleJob)

Alternatively, you can use the following options to cancel an interactive job:

(1) The DolphinDB GUI provides a button for canceling the job currently running in GUI. If the GUI is forced to quit when a job is still running, a command is automatically sent to cancel the running job. (2) In the DolphinDB Web interface, select a node to view its running jobs. Click "Stop" to cancel a job.

Note:

• It may take a while to cancel an interactive job after you send the cancel command. In DolphinDB, jobs are executed based on threads. As threads cannot be directly killed (as opposed to processes), DolphinDB sets a Cancel flag on a job after receiving the cancel command, and then checks for the Cancel flag at some stage during job execution (e.g., before starting a task or before each loop iteration). If a Cancel flag is detected, the job is canceled.
• If an interactive job cannot be canceled, it's usually because jobs are queuing up on the node and cancelConsoleJob cannot get executed immediately. One solution is to use the "Cancel" button in the DolphinDB GUI menu bar. Another solution is to use the "Stop" button in the Web interface. Cancel commands submitted using these two buttons are marked as urgent and handled with top priority. You can also cancel the job using DolphinDB console (CLI) - jobs submitted via the console will be executed immediately without entering a queue to wait.

A batch job is submitted with submitJob or submitJobEx. The difference is that submitJobEx enables you to specify job priority and parallelism. The concepts of priority and parallelism are explained in detail later in this tutorial.

For time-consuming tasks, such as writing historical data to a DFS database and ad hoc queries, we can encapsulate them into a function and submit the function as a batch job. Separated from the interactive jobs, batch jobs are executed in an independent pool of workers (known as batch job workers). The maximum number of batch job workers is set by the configuration parameter maxBatchJobWorker. If the number of batch jobs exceeds the upper limit, the incoming batch jobs will enter a queue. Batch job workers are automatically destroyed after being idle for more than 60 seconds.

Batch jobs are submitted to the node currently connected to the client. To submit a batch job to another node in the cluster, call the rpc function. For example, the following script defines the function demoJob and submits it to the remote node "node2" for execution.

def demoJob(n){
	s = 0
	for (x in 1 : n) {
		s += sum(sin rand(1.0, 100000000)-0.5)
	}
	return s
};
rpc("node2", submitJob, "demoJob", "a demo job", demoJob, 1000);

Call getRecentJobs to view the details about the batch jobs on the local node. This example retrieves the most recent 4 batch jobs:

getRecentJobs(4);
//output
node	userID	jobId   jobDesc	priority    parallelism	receivedTime	        startTime               endTime                 errorMsg
----    ------  -----   ------- --------    ----------- ------------            ---------               -------                 --------
NODE_0	admin	write   write	4	    2	        2020.03.08T16:09:16.795			
NODE_0	admin	write   check	4	    2	        2020.01.08T16:09:16.795	2020.01.08T16:09:16.797		
NODE_0	guest	query   query	0	    1	        2020.01.10T21:44:16.122	2020.01.10T21:44:16.123	2020.01.10T21:44:16.123	Not granted to read table dfs://FuturesContract/
NODE_0	admin	test    foo     8	    64	        2020.02.25T01:30:23.458	2020.02.25T01:30:23.460	2020.02.25T01:30:23.460	

In the above table:

• If startTime is empty (e.g., the first record), it means the job is still pending execution.
• If endTime is empty (e.g., the second record), it means the job is in execution.
• If errorMsg is NOT empty (e.g., the third record), it means an error occurred during the execution.

getRecentJobs returns information on the recently submitted batch jobs. To view the information on a specific job, use getJobStatus by specifying the job ID. To view the batch jobs on other nodes, use rpc or pnodeRun. The following script retrieves the batch jobs on all data nodes and compute nodes in a cluster:

  pnodeRun(getRecentJobs)

Note: getRecentJobs only returns jobs submitted after the current node was started. It does not return jobs submitted before the server was rebooted. To view jobs submitted prior to the last reboot, check the job files persisted on disk storage.

DolphinDB saves the output of batch jobs to disk files whose directory is specified by the configuration parameter batchJobDir. If the parameter is not specified, the default path is <HomeDir>/batchJobs. Each batch job generates 2 files: <job_id>.msg and <job_id>.obj that store intermediate messages and return objects, respectively. In addition, each batch job adds 3 entries to the batch job log <BatchJobDir>/batchJob.log when the system receives, starts, and completes the batch jobs. Use the following functions to check the execution of a batch job:

• getJobMessage: Retrieve the intermediate messages from the batch job.
• getJobReturn: Retrieve the batch job result.

Note: You can also use the DolphinDB Web interface to view the batch jobs. See section "View Interactive Jobs".

Use cancelJob to cancel submitted but unfinished batch jobs. As with the interactive jobs, the cancel command doesn't take effect immediately (see section 1.3 Cancel Interactive Jobs).

To cancel batch jobs on other nodes, use rpc or pnodeRun. The following script cancels the batch jobs on all data nodes and compute nodes in a cluster:

def cancelAllBatchJob(){
	jobids=exec jobid from getRecentJobs() where endTime=NULL
	cancelJob(jobids)	
}
pnodeRun(cancelAllBatchJob)

The workers and executors are further divided into the following categories:

• worker: A thread that handles regular interactive tasks. A worker receives requests from the client and breaks down each request into tasks. Based on the locations of the involved partitions in the tasks, the worker assigns the tasks to local and/or remote executors. The worker also executes local tasks itself. The number of workers is specified by the configuration parameter workerNum and the default value is the number of CPU cores. Since version 1.30.19/2.00.7, DolphinDB has introduced a multilevel task algorithm which divides the workers into different levels to handle the jobs/tasks at the same level. This helps to prevent deadlock during the execution of a particularly complex job that will be eventually broken down into several levels of subtasks.
• local executor: A thread that executes tasks on local machines assigned by workers. A worker breaks down a job into a number of tasks and adds them to the task queues. Each local executor can only process one task from the local task queue at a time. Computational tasks generated by the parallel-processing functions such as ploop and peach are also handled by the local executors. The number of local executor is specified by the configuration parameter localExecutors and the default value is the number of CPU cores - 1.

Note: The concurrent processing capability of DolphinDB is determined by the maximum number of workers and local executors.

• remote executor: An independent thread that sends tasks in the remote task queue to the remote nodes.
• batch job worker: A thread that handles batch jobs. The number of batch job workers is specified by the configuration parameter maxBatchJobWorker and the default value is equivalent to the value of workerNum. A batch job worker will be destroyed after being idle for over 60 seconds.
• dynamic worker: When all workers are busy, the system will create dynamic workers to process incoming jobs. Depending on how busy the system is with concurrent tasks, a total of three groups of dynamic workers can be created with maxDynamicWorker dynamic threads in each group. A dynamic worker will be destroyed after being idle for 60 seconds.
• web worker: A thread that handles HTTP requests. DolphinDB provides the Web-based Cluster Manager to interact with cluster nodes. Requests submitted through the web interface are processed by the web workers. The number of web workers is specified by the configuration parameter webWorkerNum and the default value is 1.
• urgent worker: A thread that only handles the time-sensitive system-level tasks such as logins (login) and job cancellations (cancelJob/cancelConsoleJob).
• infra worker: When high availability is enabled for controllers or streaming, the system will create infra workers for data communication between nodes in the raft group.
• recovery worker: Workers that are used to recover chunks synchronously in node recovery. When the chunk data on a node is corrupted, it can be recovered from the replica on another node. The number of recovery workers is specified by the configuration parameter recoveryWorkers and the default value is 1.
• diskIO worker: Set by the configuration parameter diskIOConcurrencyLevel. diskIOConcurrencyLevel = 0 means the threads handling the current tasks will read and write to the disk directly. When diskIOConcurrencyLevel > 0, the system will create the specified number of threads to read and write to the disk concurrently. For more information about diskIOConcurrencyLevel, see section 5 "Configuration Parameters".

The following example explains the differences between workers and executors. In this example, the server is deployed on a single machine.

When a node receives a query from a non-web client such as the GUI, API, console, or VS Code plugin, it assigns the query to one of its workers at level 0. Suppose that this query involves 16 partitions in a table, and the number of workers is 8 and the number of available local executors is 7 on the node. The dedicated worker will then process 2 partitions and the local executors will process 14 partitions. After all partitions are processed, the worker collects the intermediate results, generates the final result, and sends it to the client. If the query is sent from a web client, it would be handled by a web worker in the first place instead of a worker. The rest of the process is the same.

In the above example, 8 workers are usually enough to handle all the jobs sent to a node. However, the number of workers may be inadequate if dozens of queries are submitted to a node at the same time. That's when the dynamic workers are created to help processing the concurrent queries. The dynamic workers will be automatically destroyed after being idle for 60 seconds.

Compared with the regular queries, some jobs cost significantly more time, such as batch processing for machine learning. If these two kinds of jobs share a thread pool, it is likely that the workers will be occupied by the time-consuming jobs while the regular queries will never, or have to wait for a long time to get executed. Therefore, we recommend submitting time-consuming jobs as batch jobs using submitJob. The batch jobs are processed by batch job workers. Batch job workers will be automatically destroyed after being idle for 60 seconds.

When all the workers are busy and you want to cancel a job, as mentioned in section 1.3 "Cancel an Interactive Job", you can use the cancel button in the GUI or the Web interface, or send the command from a remote node. This way, the command is executed with top priority by the urgent workers, which only handles the time-sensitive system-level tasks such as logins (login) and job cancellations.

For more information, see Overview of Threading Model in DolphinDB.

In DolphinDB, jobs are executed based on priority, which ranges from 0 to 9 - the bigger the value, the higher the priority. A multi-level queue mechanism is provided to implement the job priority system. Specifically, the system maintains 10 queues, which correspond to 10 priority levels. The system always allocates thread resources to the high-priority jobs, and for jobs of the same priority, threads are allocated in a round-robin manner; only when a priority queue is empty, the jobs in the next lower priority queue are processed.

The job priority is determined by the following conditions::

• For an interactive job, its priority is min(4, the highest job priority). To specify the highest priority of the jobs submitted by a specific user, call setMaxJobPriority.
• For a batch job submitted with submitJob, its priority is 4. You can also submit a batch job with submitJobEx to specify its priority.
• For a scheduled job, the job priority is always 4.

Parallelism indicates the maximum number of threads to execute the tasks of a job on a data node at the same time. The default value is 2. To specify the parallelism of a batch job, submit it with submitJobEx .

The degree of parallelism determines the number of time slices (i.e., how many rounds of scheduling) is required to execute a job. For example, if a job with the parallelism of 2 is broken down into 100 parallel tasks, the system will only allocate 2 threads to execute these tasks, so it will take 50 rounds of scheduling to complete the entire job. To make full use of the resources, the system will perform optimization: If there are currently 16 local executors available and only one job is running, all 16 threads will be allocated to the job even if the parallelism is 2.

When there are multiple jobs, the jobs with higher priority and higher degree of parallelism get more compute resources. To prevent the low-priority jobs from waiting too long, DolphinDB would dynamically adjust the priority of a job during its execution. Specifically, when a time slice (one round of parallel execution) for a job is executed, the system would lower its priority by one level if there is another job with a lower priority. After the priority of a job reaches the lowest level, it will then be adjusted to the initial level This technique helps avoid starvation in low-priority job execution.

Here's an example: There are 4 available threads in the resource pool. Job A has a priority of 6, parallelism of 4, and 16 tasks, while job B has a priority of 4, parallelism of 2, and 2 tasks.

In the first round of scheduling: A has a higher priority than B, so all threads are allocated to A.

After the execution, A has 12 tasks left and its priority drops to 5.

In the second round of scheduling: A has the higher priority of 5, so all threads are still allocated to A.

After the execution, A has 8 tasks left and its priority drops to 4.

In the third round of scheduling: both A and B are at priority 4, so they each get 2 tasks executed.

After the execution, B is completed, A has 6 tasks left and its priority is still 4.

In this example, although the priority and parallelism of A are higher, B still gets completed first.