OlapTableSink::send is low efficient?

[vagetablechicken]

Problem
When we use broker load, OlapTableSink::send() takes the longest time, almost all of the plan_fragment active time.
Example in one BE:

Fragment f59d832368a84c94-be109f903cf4698d:(Active: 3h36m, % non-child: 0.00%)
   - AverageThreadTokens: 1.00
   - PeakReservation: 0
   - PeakUsedReservation: 0
   - RowsProduced: 168.61M
   - SizeProduced: 30.25 GB
  BlockMgr:
     - BlockWritesOutstanding: 0
     - BlocksCreated: 0
     - BlocksRecycled: 0
     - BufferedPins: 0
     - BytesWritten: 0
     - MaxBlockSize: 8.00 MB
     - MemoryLimit: 2.00 GB
     - TotalBufferWaitTime: 0.000ns
     - TotalEncryptionTime: 0.000ns
     - TotalIntegrityCheckTime: 0.000ns
     - TotalReadBlockTime: 0.000ns
  OlapTableSink:(Active: 3h35m, % non-child: 0.00%)
     - CloseTime: 102.932ms
     - ConvertBatchTime: 0.000ns
     - OpenTime: 247.194ms
     - RowsFiltered: 0
     - RowsRead: 168.61M
     - RowsReturned: 168.61M
     - SendDataTime: 3h34m
     - SerializeBatchTime: 8m26s
     - ValidateDataTime: 19s554ms
     - WaitInFlightPacketTime: 3h23m
  BROKER_SCAN_NODE (id=0):(Active: 1m8s, % non-child: 0.00%)
     - BytesRead: 0
     - MemoryUsed: 0
     - NumThread: 0
     - PerReadThreadRawHdfsThroughput: 0.00 /sec
     - RowsRead: 168.61M
     - RowsReturned: 168.61M
     - RowsReturnedRate: 2.48 M/sec
     - ScanRangesComplete: 0
     - ScannerThreadsInvoluntaryContextSwitches: 0
     - ScannerThreadsTotalWallClockTime: 0.000ns
       - MaterializeTupleTime(*): 5m37s
       - ScannerThreadsSysTime: 0.000ns
       - ScannerThreadsUserTime: 0.000ns
     - ScannerThreadsVoluntaryContextSwitches: 0
     - TotalRawReadTime(*): 38m58s
     - TotalReadThroughput: 0.00 /sec
     - WaitScannerTime: 1m7s

As can be seen above, WaitInFlightPacketTime is the most time-consuming portion.

Analysis
I describe the whole progress here.

PlanFragmentExecutor pseudo code:

while(1){
    batch=get_one_batch();
    OlapTableSink::send(batch);
}

Then, OlapTableSink::send() pseudo code:

for(row in batch){
    channel=get_corresponding_channel(row);

    // channel::add_row() explanation:
    ok=channel::add_row_in_cur_batch(row);
    if(!ok){
        if(channel::has_in_flight_packet){
            channel::wait_in_flight_packet(); // (*)
        }
        channel::send_add_batch_req();
        channel::add_row_in_cur_batch(row);
    }
    // channel::add_row() end
}

So if we trigger channel::wait_in_flight_packet(), it will block the whole process. But there's no need to block other channels add_row().
For example, channel0 is waiting in_flight_packet, we can still add row to other channels.

Better solutions(preliminary thoughts)

• make channel::add_row() non-blocking. It might be a massive change.
• make channel::add_row() less blocking. e.g. avoid adding row to channel0 immediately after channel0 send a add_batch request.

[morningman]

Yes, you are right. Currently, there is only one thread to do add rows -> send data -> wait in flight. And actually, add rows and send data can be done paralleled.

A simple implementation is to use one thread to add rows, and a thread pool to do the data sending. But may double the memory consumption cause it may has 2 row batch in memory
same time.

[vagetablechicken]

@morningman
That means the memory usage of a row batch may be great? So thus, we should stop pushing more batch as follows:
https://github.com/apache/incubator-doris/blob/8df63bc191fd123df5986084c1e453db732fda29/be/src/exec/broker_scan_node.cpp#L366-L378

I think even if in the single-thread model, it's more than 1 row batch in memory.
Let's consider the worst-case scenarios.

The single-thread model

Divide into three roles:

1 scanner (holds one _batch_queue):

The mem usage of _batch_queue is denoted by Size_queue.
_batch_queue is not empty and memory exceed limit(aka mem_limit), thus

Size_queue > mem_limit

1 plan_fragment_executor (holds one _row_batch)：
_row_batch comes from _batch_queue, denoted by Size_batch. If it's too large, we can assume:

Size_batch > mem_limit

N_nc NodeChannels(holds deep_copy rows, aka "_batch")

If "_batch"s in all channels are near full capacity, every channel has batch_size-1 rows, the mem usage is approximately equal to Size_batch.
The mem usage of all NodeChannels is N_nc*Size_batch.

So, Mem_worst = Size_queue+(1+N_nc)*Size_batch.

Mem_worst > (2+N_nc)*mem_limit.

The multi-thread model

If we use a thread pool or whatever to do data sending, we should add a new role, sender. The best parallelism is that one sender is in charge of one channel.

1 scanner (holds one _batch_queue) SAME

1 plan_fragment_executor (holds one _row_batch) SAME

N_nc send_worker
send_workers will buffer rows, this should be strictly controlled of mem usage.
The mem usage of one sender is denoted by Size_buf.
The sum is N_nc*Size_buf.

N_ch NodeChannels(holds deep_copy rows, aka "_batch") SAME

So, Mem_worst = Size_queue+(1+N_nc)*Size_batch+N_nc*Size_buf.
Mem_worst > (2+N_nc)*mem_limit + N_nc*Size_buf

If Size_buf is significantly smaller than mem_limit, it can be ignored.
If Size_buf is smaller than the bytes per row, it still might bring some benefits, cause it improves the parallel degree of wait_in_flight_packet.

[vagetablechicken]

OlapTableSink Multithreading Solution Design

single-thread model (original model)

multi-thread model

When we create an OlapTableSink, prepare N buffers and N send threads.

If a row needs to be added to NodeChannel(node_id=A, only consider node_id), the main thread copy it to buffer A%N.

We can limit the buffer size(use MemTracker), the mem_limit is configurable. If a buffer is full, we block adding rows to this buffer , until all rows in buffer have been consumed.

Extra cost of multi-thread ver

thread: buffer_num
mem: buffer_num*(mem_limit+buffer_running_mem)

single/multi version switch

The acceleration effect is evident for large data imports. But no need for every OlapTableSink.
So I design it:
We can select origin(single-thread) version or multi-thread version by set broker load property, and configure the buffer_num & mem_limit_per_buffer.
We add fields to TOlapTableSink, like

struct TOlapTableSink {
    ...
    14: optional i64 load_channel_timeout_s // the timeout of load channels in second
    15: optional i32 buffer_num
    16: optional i64 mem_limit_per_buf
    17: optional i64 size_limit_per_buf
}

buffer_num > 0 means using "multi-thread & buffer" mode.
buffer_num = 0 or not set means using "single-thread" mode(the origin mode).
(buffer_num = 1 is hard to define. We should avoid setting buffer_num to 1.)

So we can use

LOAD LABEL ...
        PROPERTIES 
        ( 
        "buffer_num"="5", 
        "mem_limit_per_buf"="5368709120", 
        "size_limit_per_buf"="62914560"
        );

A Test Case

cluster: 5 be
{"ScannedRows":895737370,"TaskNumber":1,"FileNumber":300,"FileSize":60216822679} // 56G

origin ver

LoadStartTime	LoadFinishTime
2020-02-19 20:44:00	2020-02-20 00:35:01

multi-thread ver

buffer_num/mem_limit_per_buf/size_limit_per_buf = 5/1G/30M

LoadStartTime	LoadFinishTime
2020-02-18 17:57:38	2020-02-18 20:05:13

buffer_num/mem_limit_per_buf/size_limit_per_buf = 5/5G/30M

LoadStartTime	LoadFinishTime
2020-02-18 20:14:58	2020-02-18 22:19:54

buffer_num/mem_limit_per_buf/size_limit_per_buf = 5/5G/60M

LoadStartTime	LoadFinishTime
2020-02-18 22:28:08	2020-02-19 00:36:28

Test Analysis

56G broker load, the origin ver will run for 4 hours, the multi-ver can halve the time.
But increasing mem_limit & size_limit doesn't do much good.(not bad, cause we can only use minimum mem, the more memory is waste. 😀 )

1. Because we use modular hashing(node_id%buffer_num), the buffer_num should <= be_num.
   buffer_num=be_num is ideal.
2. In single-thread ver, we only have a thread to add_batch/wait_packet. In multi-thread ver, we have buffer_num thread to do them. But the add_batch/wait_packet in one thread are in sequence, can't save time in one thread. Thus, the actual consume time is the ideal of load run time. For example,
   mem_handle: 3506.55s, deep_copy: 22.1017s, spsc push block if full: 7.81192s, consume: 7298.54s, actual consume: 7285s. means the ideal run time is 7285s, our configuration can make the process 7298s. It means little to further optimize.
3. we use spsc_queue(ringbuffer) to be the buffer, it has fixed length(size_limit_per_buf). So, the size_limit_per_buf should not be too big. We can use size_limit_per_buf*30B to estimate the mem of queue.

[chaoyli]

The enhancement of this test case seems not obvious?

[vagetablechicken]

The enhancement of this test case seems not obvious?

the origin ver is being tested. It typically takes 5h. I haven't tested it recently, so I ran this test again.

[vagetablechicken]

#2956 (comment)
As mentioned, here‘s the new design of OlapTableSink--add one sender thread to do non-blocking sending.
Let me explain the meaning of non-blocking.

The origin version of OlapTableSink can be abstracted as one queue(contains all batches in all node channels). One thread consumes the queue's items, one by one. When it wants to send a batch of the NodeChannel which has a in flight packet(rpc hasn’t returned response), it must wait(rpc join).
For example:
The batch which index id=0 && node id=2, is denoted by "B1(0-2)".
The Abstract queue as shown below.

B0(0-1)	B1(0-2)	B2(0-1)	B3(1-4)	...	...

When we are sending B2(0-1), we must wait for the B0 response. But if we set aside B2(0-1) and send the next item "B3(1-4)", it won't be blocked.

So I used to split the one queue into multi queues(abandoned), as follows.( For details, see #2780 (comment))
Batches queue0:

B0(0-0)	B1(0-0)	B2(2-0)	B3(4-0)	...	...

Batches queue1:

B0(2-1)	B1(0-1)	B2(0-1)	B3(1-1)	...	...

Batches queue2:

B0(0-2)	B1(0-2)	B2(3-2)	B3(1-2)	...	...

Each queue needs one thread to consume items. Block time is shared by multiple queues. But it's still a block way.

The new design is non-blocking.

We can save batches in NodeChannels(pending batches), and try to send a pending batch. If the current channel has a in flight packet, we just skip sending in this round.

The implementation is coming soon.

[gaodayue]

The new non-blocking approach looks good. It should improve the overall load performance particularly in case where you have some slower receiver nodes.

BTW you should cap the total number of pending batches in order to control the memory usage.

[vagetablechicken]

The new non-blocking approach looks good. It should improve the overall load performance particularly in case where you have some slower receiver nodes.

BTW you should cap the total number of pending batches in order to control the memory usage.

There's no need to limit mem in sink node. When we create a new RowBatch in NodeChannel, we use sink node's mem_tracker. As we know, the sink node & the scan node has the same ancestor mem_tracker(query mem tracker， default 2GB), so the mem limit is a matter for scan node.