/cc @SeaRise

@windtalker: GitHub didn't allow me to request PR reviews from the following users: SeaRise.

Note that only pingcap members and repo collaborators can review this PR, and authors cannot review their own PRs.

/cc @SeaRise

That's amazing!

have on chance to be notified

have no chance to be notified

How about modifying the waitForWritable method of ExchangeWriter so that it returns WaitResult::Ready even if the tunnel is not ready, as long as the cache has not reached the limit? Then, when sink->write detects that the tunnel is not ready, it returns WaitResult::WAIT_FOR_NOTIFY. This way, when the sink is notified, it will definitely write to the mpmcqueue, preventing any hang-ups?

How about modifying the waitForWritable method of ExchangeWriter so that it returns WaitResult::Ready even if the tunnel is not ready, as long as the cache has not reached the limit? Then, when sink->write detects that the tunnel is not ready, it returns WaitResult::WAIT_FOR_NOTIFY. This way, when the sink is notified, it will definitely write to the mpmcqueue, preventing any hang-ups?

BY "even if the tunnel is not ready" you mean the tunnel is not connected or the tunnel is full?

How about modifying the waitForWritable method of ExchangeWriter so that it returns WaitResult::Ready even if the tunnel is not ready, as long as the cache has not reached the limit? Then, when sink->write detects that the tunnel is not ready, it returns WaitResult::WAIT_FOR_NOTIFY. This way, when the sink is notified, it will definitely write to the mpmcqueue, preventing any hang-ups?

BY "even if the tunnel is not ready" you mean the tunnel is not connected or the tunnel is full?

I mean tunnel is full.

Considering this case:
If it's hash-partitioned, and the concurrency is 2, both operators need to flush, and the number of tunnels is 2:

1. Operator1 and Operator2 are both waiting for a notification on tunnel[1].
2. Operator1 is notified by tunnel[1], and since it has data to flush, it won't call triggerPipelineWriterNotify.
3. When Operator1 flushes, just because Partition 1 has no data written, and only Partition 0 has data, tunnel[1] will not trigger a write at this time.
4. Because tunnel[1] has no data written to it, there is no consumer to consume from tunnel[1] to trigger Operator2, resulting in a hang.

So query will hang, right?

1. iting for a notifi

Current pr makes sure that operator will send data to all tunnels, even if there is no actually data to send.

Another possible fix from @SeaRise
Basic idea: currently, ExchangeSenderSinkOp will first write data to writer's internal cache, and if cache if full, then flush the cache to tunnelset, so another fix should be "only register ExchangeSenderSinkOp to pipeline notify furture if the writer wants to flush data to tunnelset".
Changes need to made:

1. call tunnel's waitForWritable only when writer wants to flush data to tunnelset, currently, we call tunnelset's waitForWritable in ExchangeSenderSinkOp->prepare(), we need to move this code inside each writer's write function, the code should be something like this

function write()
{
     // write data to cache
    if (cache is full)
    {
        if (tunnelset is writable)
            // flush data to tunnelset
        else
            // return wait_for_notify or wait_for_polling
    }
}

2. Inside ExchangeSenderSinkOp->prepare(), it should check if current writer has remaining data to flush to tunnel, the code should be something like this

function prepare() 
{
    if (writer->hasDataToFlush())
    {
        if (tunnel is writable)
            // flush data to tunnelset
        else
            // return wait_for_notify or wait_for_polling
    }
    else
    {
        // return need_input status
    }
}

3. Usually, a tunnelset contains more than 1 tunnel. Unfortunately, there is no atomic way to check if all the tunnels are writable, so in currently code, we use force_push once all tunnels claim they are ok to write independently. In this fix, it should be re-considered that if we still need to use force_push to do this or we can maintain flush data to tunnel independently for all the tunnels in the same tunnelset

Another possible fix from @SeaRise Basic idea: currently, ExchangeSenderSinkOp will first write data to writer's internal cache, and if cache if full, then flush the cache to tunnelset, so another fix should be "only register ExchangeSenderSinkOp to pipeline notify furture if the writer wants to flush data to tunnelset". Changes need to made:

1. call tunnel's waitForWritable only when writer wants to flush data to tunnelset, currently, we call tunnelset's waitForWritable in ExchangeSenderSinkOp->prepare(), we need to move this code inside each writer's write function, the code should be something like this

function write()
{
     // write data to cache
    if (cache is full)
    {
        if (tunnelset is writable)
            // flush data to tunnelset
        else
            // return wait_for_notify or wait_for_polling
    }
}

2. Inside ExchangeSenderSinkOp->prepare(), it should check if current writer has remaining data to flush to tunnel, the code should be something like this

function prepare() 
{
    if (writer->hasDataToFlush())
    {
        if (tunnel is writable)
            // flush data to tunnelset
        else
            // return wait_for_notify or wait_for_polling
    }
    else
    {
        // return need_input status
    }
}

3. Usually, a tunnelset contains more than 1 tunnel. Unfortunately, there is no atomic way to check if all the tunnels are writable, so in currently code, we use force_push once all tunnels claim they are ok to write independently. In this fix, it should be re-considered that if we still need to use force_push to do this or we can maintain flush data to tunnel independently for all the tunnels in the same tunnelset

After some offline discussion, we decide to use this pr as a quick fix for the hang issue, and will refine the whole tunnel write process using the above idea.

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: gengliqi, SeaRise

The full list of commands accepted by this bot can be found here.

The pull request process is described here

[LGTM Timeline notifier]

Timeline:

• 2024-09-10 10:25:10.100761417 +0000 UTC m=+351979.841185357: ☑️ agreed by gengliqi.
• 2024-09-10 12:45:53.321848502 +0000 UTC m=+360423.062272441: ☑️ agreed by SeaRise.