Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1pDgKd-00011d-Ep for pgsql-hackers@arkaria.postgresql.org; Fri, 06 Jan 2023 06:35:59 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.92) (envelope-from ) id 1pDgKc-0002BJ-BT for pgsql-hackers@arkaria.postgresql.org; Fri, 06 Jan 2023 06:35:58 +0000 Received: from makus.postgresql.org ([2001:4800:3e1:1::229]) by malur.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1pDgKb-0002B4-S4 for pgsql-hackers@lists.postgresql.org; Fri, 06 Jan 2023 06:35:58 +0000 Received: from mail-oi1-x229.google.com ([2607:f8b0:4864:20::229]) by makus.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_128_GCM_SHA256:128) (Exim 4.92) (envelope-from ) id 1pDgKZ-0005pK-B7 for pgsql-hackers@lists.postgresql.org; Fri, 06 Jan 2023 06:35:56 +0000 Received: by mail-oi1-x229.google.com with SMTP id s187so418024oie.10 for ; Thu, 05 Jan 2023 22:35:55 -0800 (PST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20210112; h=cc:to:subject:message-id:date:from:in-reply-to:references :mime-version:from:to:cc:subject:date:message-id:reply-to; bh=hxFMOZjMrVBCYd3UiMvNy8AFWj8sBXADMITvEMdQXhU=; b=JxtjOjpgRzYsIsfhGxTjH6HumVjgi3Td1Jkx+IsVg6HtmEf2vB0CIio99v5I/AloWP RT5XKu1z/0tOcJuY5yYkzcomyts7C/77qFdlLhxWhFl3VMj7yTfURF+oxMk9bs7LOFP7 oTv3PAy4KSwotA8UPYEShyLH14OSCUEfs0xhS3x8ZojITA+onh2uA8H/JQmMMUVq6E5L 75oFCYqkYj7Rn9MqrKvKfFZ1rMi04tr9FahomfeySQxQkt6K039tS+rHhmCC+G2ozAne X5W7lBi3gDFRlv2rWz20tQMelSR1n0k9JlZ8cjCsV2gemlbWGaxt6J3DK0U0PhK5GIsM kLcw== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20210112; h=cc:to:subject:message-id:date:from:in-reply-to:references :mime-version:x-gm-message-state:from:to:cc:subject:date:message-id :reply-to; bh=hxFMOZjMrVBCYd3UiMvNy8AFWj8sBXADMITvEMdQXhU=; b=hmYTklS39GJc/2NS6fCH7NB4aELgeSz8XaX1n0zaI6hkbKA4V/OZiY46NS2ILpwJKk fwcw/wDwAotk7sb7r7mni0IFUy62ZzUrgnPJPjBTgdnRkWx7NkPaV1wSb60NsN61g7BH LBF6sBHs0KelL62xH5V5Ximgx502n/wVBMiuKsFCluHdh03DD/t0w2EhQ97QmHMjSLPF cBvHtCKgGzDiq4/X5j605WaXvfaNmEgMoGqFIV9Rw9/KCtHabIWRwcsg5TXYwrPYeIY6 RfSyYDwR/xI3y1S55yJNErC4pKrXwKg9B1k0jBN2MpFqvGVJL6z9HvudiOBe5E1siTTJ MNlg== X-Gm-Message-State: AFqh2koZpTC6TKtyPt3EQPrDZeXptHNmoqdrH/iyAO4Dp1L0SvRq/pk0 laZIJ24/1Fyk93wFKMQ1laN5bXhbRErXZanWxkQ= X-Google-Smtp-Source: AMrXdXuzARZvO0OUyRF+hp5+lJR2CTqOktO/PJfdvS9cQsrGICSQf1Wn+5IKQMMJCrN3cTFBw6EBw0L+9KHcHOIK5pg= X-Received: by 2002:a54:4404:0:b0:35a:8ab0:70f0 with SMTP id k4-20020a544404000000b0035a8ab070f0mr3169923oiw.272.1672986954431; Thu, 05 Jan 2023 22:35:54 -0800 (PST) MIME-Version: 1.0 References: In-Reply-To: From: Amit Kapila Date: Fri, 6 Jan 2023 12:05:42 +0530 Message-ID: Subject: Re: Perform streaming logical transactions by background workers and parallel apply To: Dilip Kumar Cc: "houzj.fnst@fujitsu.com" , Masahiko Sawada , "wangw.fnst@fujitsu.com" , Peter Smith , "shiy.fnst@fujitsu.com" , PostgreSQL Hackers Content-Type: text/plain; charset="UTF-8" List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Archived-At: Precedence: bulk On Fri, Jan 6, 2023 at 11:24 AM Dilip Kumar wrote: > > On Fri, Jan 6, 2023 at 9:37 AM houzj.fnst@fujitsu.com > wrote: > > > > On Thursday, January 5, 2023 7:54 PM Dilip Kumar wrote: > > Thanks, I have started another thread[1] > > > > Attach the parallel apply patch set here again. I didn't change the patch set, > > attach it here just to let the CFbot keep testing it. > > I have completed the review and some basic testing and it mostly looks > fine to me. Here is my last set of comments/suggestions. > > 1. > /* > * Don't start a new parallel worker if user has set skiplsn as it's > * possible that they want to skip the streaming transaction. For > * streaming transactions, we need to serialize the transaction to a file > * so that we can get the last LSN of the transaction to judge whether to > * skip before starting to apply the change. > */ > if (!XLogRecPtrIsInvalid(MySubscription->skiplsn)) > return false; > > > I think this is fine to block parallelism in this case, but it is also > possible to make it less restrictive, basically, only if the first lsn > of the transaction is <= skiplsn, then only it is possible that the > final_lsn might match with skiplsn otherwise that is not possible. And > if we want then we can allow parallelism in that case. > > I understand that currently we do not have first_lsn of the > transaction in stream start message but I think that should be easy to > do? Although I am not sure if it is worth it, it's good to make a > note at least. > Yeah, I also don't think sending extra eight bytes with stream_start message is worth it. But it is fine to mention the same in the comments. > 2. > > + * XXX Additionally, we also stop the worker if the leader apply worker > + * serialize part of the transaction data due to a send timeout. This is > + * because the message could be partially written to the queue and there > + * is no way to clean the queue other than resending the message until it > + * succeeds. Instead of trying to send the data which anyway would have > + * been serialized and then letting the parallel apply worker deal with > + * the spurious message, we stop the worker. > + */ > + if (winfo->serialize_changes || > + list_length(ParallelApplyWorkerPool) > > + (max_parallel_apply_workers_per_subscription / 2)) > > IMHO this reason (XXX Additionally, we also stop the worker if the > leader apply worker serialize part of the transaction data due to a > send timeout) for stopping the worker looks a bit hackish to me. It > may be a rare case so I am not talking about the performance but the > reasoning behind stopping is not good. Ideally we should be able to > clean up the message queue and reuse the worker. > TBH, I don't know what is the better way to deal with this with the current infrastructure. I thought we can do this as a separate enhancement in the future. > 3. > + else if (shmq_res == SHM_MQ_WOULD_BLOCK) > + { > + /* Replay the changes from the file, if any. */ > + if (pa_has_spooled_message_pending()) > + { > + pa_spooled_messages(); > + } > > I think we do not need this pa_has_spooled_message_pending() function. > Because this function is just calling pa_get_fileset_state() which is > acquiring mutex and getting filestate then if the filestate is not > FS_EMPTY then we call pa_spooled_messages() that will again call > pa_get_fileset_state() which will again acquire mutex. I think when > the state is FS_SERIALIZE_IN_PROGRESS it will frequently call > pa_get_fileset_state() consecutively 2 times, and I think we can > easily achieve the same behavior with just one call. > This is just to keep the code easy to follow. As this would be a rare case, so thought of giving preference to code clarity. -- With Regards, Amit Kapila.