Latency Measurement and Performance Improvement of Stream Computing

Suppose we receive real-time tick-by-tick transaction data and respond to each transaction record individually, accumulating the latest daily total transaction volume. If each calculation uses all transaction data up to the current point, performance would be poor. However, through incremental streaming implementation, performance can be greatly improved. Specifically, the latest cumulative transaction volume can be obtained by adding the transaction volume of the latest record to the previously calculated transaction volume. This incremental calculation requires historical state (previously calculated transaction volume), which we call stateful computation.

DolphinDB provides numerous built-in state functions in reactive state engines, time-series aggregation engines, and window join engines. Using built-in state functions in metrics can implement the above incremental stateful computation, with historical states automatically maintained by the engine internally. For example, using the cumsum function in reactive state engines can implement incremental accumulation.

For state functions supported by each engine, refer to the user manual for createTimeSeriesEngine, createReactiveStateEngine, and createWindowJoinEngine. It's recommended to prioritize built-in state functions for implementing real-time indicators with incremental algorithms.

Just-in-time compilation is a form of dynamic compilation that can improve program execution efficiency. DolphinDB’s scripting language is interpreted. When a program runs, it first performs syntax analysis to generate a syntax tree, and then executes it recursively. Without vectorization, the interpretation overhead can be relatively high, because DolphinDB is implemented in C++, and a single function call in the script may translate into multiple virtual function calls in C++.

Since streaming computation tasks are continuously triggered, functions are repeatedly invoked. For example, as described in Section 1.3 on reactive state engines, injecting just 10 records will result in 5 function calls for the metrics. The interpretation overhead accumulates and contributes to the overall processing latency. This is especially significant for reactive state engines, where in some scenarios, a large portion of the total latency may come from interpretation. Therefore, it is recommended to reduce interpretation overhead by implementing JIT functions. For practical examples, please refer to Stream Processing of Financial Factors, which demonstrates performance improvements for a moving average buy-sell pressure factor using JIT.

In DolphinDB, an array vector is a special type of vector used to store variable-length two-dimensional arrays. This data structure can significantly simplify certain common queries and calculations. For details, please refer to Best Practices for Array Vectors.

The level-2 order book (10 best bid/ask prices and volumes) is one of the most important components in market snapshots. When developing high-frequency factors, vectorized computing on this data is often desirable. The fixedLengthArrayVector function can be used to combine 10 best bid/ask data into an array vector. Leveraging this feature, it is recommended to store these fields directly as array vectors. This eliminates the need to assemble the 10 best bid/ask data during function execution, thereby reducing computational latency. For practical examples, please refer to Stream Processing of Financial Factors, which demonstrates how array vectors can be used to optimize the performance of a moving average buy-sell pressure factor.

Not all calculations require streaming engines. Simple stateless computations can often be handled directly in custom functions. For example, to process only trades after 09:30, you can filter data in a custom handler before sending it to the engine:

def filterTime(msg){
	tmp = select * from msg where time > 09:30:00.000
	getStreamEngine("reactiveDemo").append!(tmp) 
}
subscribeTable(tableName="tickStream", actionName="demo", offset=-1, handler=filterTime, msgAsTable=true)

• The same filtering can be achieved by adding another layer of reactive state engine, but due to overhead from grouping and state maintenance within streaming engines, custom functions are more efficient implementations than streaming engines for stateless computation scenarios.
• The handler function processes incremental data each time, meaning the msg variable in the above script is newly added data received by the subscriber, not the latest full snapshot of the tickStream table.

By setting appropriate batchSize and throttle parameters when calling subscribeTable, micro-batch processing can be achieved to improve throughput, reduce blocking, and thus lower latency. This leverages the fact that per-batch processing time does not increase linearly with batch size. For example, writing 1000 records to a distributed database may take almost the same time as writing 10 records. Therefore, utilizing batch processing can significantly reduce overall latency.

The batchSize and throttle parameters jointly determine consumption triggers, and are evaluated with an OR condition. When the number of unprocessed records in the subscription queue reaches batchSize, consumption is triggered. However, due to possible fluctuations in data arrival rates, there may be times when batchSize is not reached for a while. In such cases, throttle ensures that data accumulated for a specified time is still processed. Both parameters should be adjusted based on data input rates and processing speeds.

Additionally, note the following two points:

• The system configuration subThrottle (in dolphindb.cfg or cluster.cfg) must be set to 1 to allow throttle values smaller than 1 second. Otherwise, even if subscribeTable specifies throttle as 0.001, the effective value remains 1 second.
• It is recommended to always specify both batchSize and throttle when calling subscribeTable. If not specified, each batch processed corresponds to the size of data blocks as they arrive in the queue, not the total accumulated data. For example, if 10 unprocessed records arrive in blocks of 2, 3, and 5 records, the worker will process them in three separate batches, rather than one batch of 10 records. Therefore, if records are inserted frequently but in small quantities, failing to configure batchSize and throttle may result in poor performance.

The hash parameter of subscribeTable can be configured to improve parallelism and reduce latency. Each call to subscribeTable allocates a fixed message processing thread to its handler function. Proper thread allocation can maximize CPU utilization. Optimization strategies include:

• Assign different streaming tasks to different processing threads:
  • Whenever possible, assign different tasks to separate threads. If multiple streaming tasks share the same hash value, they are allocated to the same thread and compete for resources, increasing response latency for each task.
  • When the number of tasks exceeds available processing threads, prioritize assigning complex (long-running) tasks to separate threads.
• Split a single streaming task into multiple threads:
  • For high-volume input streams that cannot be handled by a single thread, data can be partitioned across multiple threads for parallel processing. See Stream for DolphinDB for implementation details. Note that writing results from multiple threads into a shared memory table may involve write locks and waiting. To avoid this, results can be written into separate tables.
  • dispatchStreamEngine offers a lightweight alternative to subscribeTable for distributing data and parallel processing.

In DolphinDB, complex indicators are often decomposed into multiple stages, each handled by a separate engine. Traditionally, intermediate results are stored in intermediate tables, with subsequent engines consuming these results via additional subscriptions. However, this introduces memory and latency overhead. DolphinDB stream engines support the table interface, allowing one engine to directly feed another without intermediate tables—a technique known as stream engine cascading—which offers better performance.

While splitting a task into multiple parallel threads via multiple subscribeTable subscriptions can improve throughput, having too many subscriptions may create publishing bottlenecks since there is only one publisher thread per node. In versions 1.30.22, 2.00.9 and above, DolphinDB introduced the Stream Dispatch Engine (createStreamDispatchEngine), which distributes incoming data across multiple threads and directly injects incremental data into downstream output tables or engines on those threads.

In the following example, data is subscribed once, then written into a dispatch engine. The dispatch engine partitions data by the sym field across three threads for computation using reactive state engines.

//  create engine
share streamTable(1:0, `sym`price, [STRING,DOUBLE]) as tickStream
share streamTable(1000:0, `sym`factor1, [STRING,DOUBLE]) as resultStream
rseArr = array(ANY, 0)
for(i in 0..2){
	rse = createReactiveStateEngine(name="reactiveDemo"+string(i), metrics =<cumavg(price)>, dummyTable=tickStream, outputTable=resultStream, keyColumn="sym")
	rseArr.append!(rse)
}
dispatchEngine=createStreamDispatchEngine(name="dispatchDemo", dummyTable=tickStream, keyColumn=`sym, outputTable=rseArr, mode="buffer", dispatchType="uniform")
//  subscribe
subscribeTable(tableName="tickStream", actionName="dispatch", handler=dispatchEngine, msgAsTable=true)

How Stream Dispatch Engine Works:

• The outputTable parameter is a list of tables (or engines). When created, an equal number of threads and buffer queues are allocated.
• Each thread processes data from its buffer queue and writes it into the corresponding output table or engine. If dispatching into engines, the calculations are performed directly on these threads.
• These threads are independent of the subscribeTable message processing threads and are not constrained by the subExecutor parameter in system configuration.

Best Practices for Stream Dispatch Engine:

• It is recommended to use the default mode"buffer", which processes all unprocessed data in batches — consistent with micro-batch processing best practices.
• When using dispatchType="hash", skewed data distributions may overload some threads. In such cases, use dispatchType="uniform" for even distribution:
  • "hash" works like hashBucket function, where each key is consistently mapped to a fixed bucket.
  • "uniform" dynamically assigns keys in arrival order, distributing keys evenly across buckets.
• The number of dispatch engine threads should not exceed available logical CPU cores to avoid excessive thread switching overhead.