Hi, I'm not sure. Maybe 5 will be fine? Anton, Dawid, what do you think?
Can you create a parent ticket for the whole FLIP to group all of the issues together? Also FLIP should be officially voted first. Best, Piotrek pt., 6 maj 2022 o 09:08 rui fan <1996fan...@gmail.com> napisał(a): > Hi Anton, Piotrek and Dawid, > > Thanks for your help. > > I created FLINK-27522[1] as the first task. And I will finish it asap. > > @Piotrek, for the default value, do you think it should be less > than 5? What do you think about 3? Actually, I think 5 isn't big. > It's 1 or 3 or 5 that doesn't matter much, the focus is on > reasonably resolving deadlock problems. Or I push the second > task to move forward first and we discuss the default value in PR. > > For the legacySource, I got your idea. And I propose we create > the third task to handle it. Because it is independent and for > compatibility with the old API. What do you think? I updated > the third task on FLIP-227[2]. > > If all is ok, I will create a JIRA for the third Task and add it to > FLIP-227. And I will develop them from the first task to the > third task. > > Thanks again for your help. > > [1] https://issues.apache.org/jira/browse/FLINK-27522 > [2] > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-227%3A+Support+overdraft+buffer > > Thanks > fanrui > > On Fri, May 6, 2022 at 3:50 AM Piotr Nowojski <pnowoj...@apache.org> > wrote: > > > Hi fanrui, > > > > > How to identify legacySource? > > > > legacy sources are always using the SourceStreamTask class and > > SourceStreamTask is used only for legacy sources. But I'm not sure how to > > enable/disable that. Adding `disableOverdraft()` call in SourceStreamTask > > would be better compared to relying on the `getAvailableFuture()` call > > (isn't it used for back pressure metric anyway?). Ideally we should > > enable/disable it in the constructors, but that might be tricky. > > > > > I prefer it to be between 5 and 10 > > > > I would vote for a smaller value because of FLINK-13203 > > > > Piotrek > > > > > > > > czw., 5 maj 2022 o 11:49 rui fan <1996fan...@gmail.com> napisał(a): > > > >> Hi, > >> > >> Thanks a lot for your discussion. > >> > >> After several discussions, I think it's clear now. I updated the > >> "Proposed Changes" of FLIP-227[1]. If I have something > >> missing, please help to add it to FLIP, or add it in the mail > >> and I can add it to FLIP. If everything is OK, I will create a > >> new JIRA for the first task, and use FLINK-26762[2] as the > >> second task. > >> > >> About the legacy source, do we set maxOverdraftBuffersPerGate=0 > >> directly? How to identify legacySource? Or could we add > >> the overdraftEnabled in LocalBufferPool? The default value > >> is false. If the getAvailableFuture is called, change > >> overdraftEnabled=true. > >> It indicates whether there are checks isAvailable elsewhere. > >> It might be more general, it can cover more cases. > >> > >> Also, I think the default value of 'max-overdraft-buffers-per-gate' > >> needs to be confirmed. I prefer it to be between 5 and 10. How > >> do you think? > >> > >> [1] > >> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-227%3A+Support+overdraft+buffer > >> [2] https://issues.apache.org/jira/browse/FLINK-26762 > >> > >> Thanks > >> fanrui > >> > >> On Thu, May 5, 2022 at 4:41 PM Piotr Nowojski <pnowoj...@apache.org> > >> wrote: > >> > >>> Hi again, > >>> > >>> After sleeping over this, if both versions (reserve and overdraft) have > >>> the same complexity, I would also prefer the overdraft. > >>> > >>> > `Integer.MAX_VALUE` as default value was my idea as well but now, as > >>> > Dawid mentioned, I think it is dangerous since it is too implicit for > >>> > the user and if the user submits one more job for the same TaskManger > >>> > >>> As I mentioned, it's not only an issue with multiple jobs. The same > >>> problem can happen with different subtasks from the same job, > potentially > >>> leading to the FLINK-13203 deadlock [1]. With FLINK-13203 fixed, I > would be > >>> in favour of Integer.MAX_VALUE to be the default value, but as it is, I > >>> think we should indeed play on the safe side and limit it. > >>> > >>> > I still don't understand how should be limited "reserve" > >>> implementation. > >>> > I mean if we have X buffers in total and the user sets overdraft > equal > >>> > to X we obviously can not reserve all buffers, but how many we are > >>> > allowed to reserve? Should it be a different configuration like > >>> > percentegeForReservedBuffers? > >>> > >>> The reserve could be defined as percentage, or as a fixed number of > >>> buffers. But yes. In normal operation subtask would not use the > reserve, as > >>> if numberOfAvailableBuffers < reserve, the output would be not > available. > >>> Only in the flatMap/timers/huge records case the reserve could be used. > >>> > >>> > 1. If the total buffers of LocalBufferPool <= the reserve buffers, > >>> will LocalBufferPool never be available? Can't process data? > >>> > >>> Of course we would need to make sure that never happens. So the reserve > >>> should be < total buffer size. > >>> > >>> > 2. If the overdraft buffer use the extra buffers, when the downstream > >>> > task inputBuffer is insufficient, it should fail to start the job, > and > >>> then > >>> > restart? When the InputBuffer is initialized, it will apply for > enough > >>> > buffers, right? > >>> > >>> The failover if downstream can not allocate buffers is already > >>> implemented FLINK-14872 [2]. There is a timeout for how long the task > is > >>> waiting for buffer allocation. However this doesn't prevent many > >>> (potentially infinitely many) deadlock/restarts cycles. IMO the propper > >>> solution for [1] would be 2b described in the ticket: > >>> > >>> > 2b. Assign extra buffers only once all of the tasks are RUNNING. This > >>> is a simplified version of 2a, without tracking the tasks > sink-to-source. > >>> > >>> But that's a pre-existing problem and I don't think we have to solve it > >>> before implementing overdraft. I think we would need to solve it only > >>> before setting Integer.MAX_VALUE as the default for the overdraft. > Maybe I > >>> would hesitate setting the overdraft to anything more then a couple of > >>> buffers by default for the same reason. > >>> > >>> > Actually, I totally agree that we don't need a lot of buffers for > >>> overdraft > >>> > >>> and > >>> > >>> > Also I agree we ignore the overdraftBuffers=numberOfSubpartitions. > >>> > When we finish this feature and after users use it, if users feedback > >>> > this issue we can discuss again. > >>> > >>> +1 > >>> > >>> Piotrek > >>> > >>> [1] https://issues.apache.org/jira/browse/FLINK-13203 > >>> [2] https://issues.apache.org/jira/browse/FLINK-14872 > >>> > >>> czw., 5 maj 2022 o 05:52 rui fan <1996fan...@gmail.com> napisał(a): > >>> > >>>> Hi everyone, > >>>> > >>>> I still have some questions. > >>>> > >>>> 1. If the total buffers of LocalBufferPool <= the reserve buffers, > will > >>>> LocalBufferPool never be available? Can't process data? > >>>> 2. If the overdraft buffer use the extra buffers, when the downstream > >>>> task inputBuffer is insufficient, it should fail to start the job, and > >>>> then > >>>> restart? When the InputBuffer is initialized, it will apply for enough > >>>> buffers, right? > >>>> > >>>> Also I agree we ignore the overdraftBuffers=numberOfSubpartitions. > >>>> When we finish this feature and after users use it, if users feedback > >>>> this issue we can discuss again. > >>>> > >>>> Thanks > >>>> fanrui > >>>> > >>>> On Wed, May 4, 2022 at 5:29 PM Dawid Wysakowicz < > dwysakow...@apache.org> > >>>> wrote: > >>>> > >>>>> Hey all, > >>>>> > >>>>> I have not replied in the thread yet, but I was following the > >>>>> discussion. > >>>>> > >>>>> Personally, I like Fanrui's and Anton's idea. As far as I understand > >>>>> it > >>>>> the idea to distinguish between inside flatMap & outside would be > >>>>> fairly > >>>>> simple, but maybe slightly indirect. The checkAvailability would > >>>>> remain > >>>>> unchanged and it is checked always between separate invocations of > the > >>>>> UDF. Therefore the overdraft buffers would not apply there. However > >>>>> once > >>>>> the pool says it is available, it means it has at least an initial > >>>>> buffer. So any additional request without checking for availability > >>>>> can > >>>>> be considered to be inside of processing a single record. This does > >>>>> not > >>>>> hold just for the LegacySource as I don't think it actually checks > for > >>>>> the availability of buffers in the LocalBufferPool. > >>>>> > >>>>> In the offline chat with Anton, we also discussed if we need a limit > >>>>> of > >>>>> the number of buffers we could overdraft (or in other words if the > >>>>> limit > >>>>> should be equal to Integer.MAX_VALUE), but personally I'd prefer to > >>>>> stay > >>>>> on the safe side and have it limited. The pool of network buffers is > >>>>> shared for the entire TaskManager, so it means it can be shared even > >>>>> across tasks of separate jobs. However, I might be just unnecessarily > >>>>> cautious here. > >>>>> > >>>>> Best, > >>>>> > >>>>> Dawid > >>>>> > >>>>> On 04/05/2022 10:54, Piotr Nowojski wrote: > >>>>> > Hi, > >>>>> > > >>>>> > Thanks for the answers. > >>>>> > > >>>>> >> we may still need to discuss whether the > >>>>> >> overdraft/reserve/spare should use extra buffers or buffers > >>>>> >> in (exclusive + floating buffers)? > >>>>> > and > >>>>> > > >>>>> >> These things resolve the different problems (at least as I see > >>>>> that). > >>>>> >> The current hardcoded "1" says that we switch "availability" to > >>>>> >> "unavailability" when one more buffer is left(actually a little > less > >>>>> >> than one buffer since we write the last piece of data to this last > >>>>> >> buffer). The overdraft feature doesn't change this logic we still > >>>>> want > >>>>> >> to switch to "unavailability" in such a way but if we are already > in > >>>>> >> "unavailability" and we want more buffers then we can take > >>>>> "overdraft > >>>>> >> number" more. So we can not avoid this hardcoded "1" since we need > >>>>> to > >>>>> >> understand when we should switch to "unavailability" > >>>>> > Ok, I see. So it seems to me that both of you have in mind to keep > >>>>> the > >>>>> > buffer pools as they are right now, but if we are in the middle of > >>>>> > processing a record, we can request extra overdraft buffers on top > of > >>>>> > those? This is another way to implement the overdraft to what I was > >>>>> > thinking. I was thinking about something like keeping the > >>>>> "overdraft" or > >>>>> > more precisely buffer "reserve" in the buffer pool. I think my > >>>>> version > >>>>> > would be easier to implement, because it is just fiddling with > >>>>> min/max > >>>>> > buffers calculation and slightly modified `checkAvailability()` > >>>>> logic. > >>>>> > > >>>>> > On the other hand what you have in mind would better utilise the > >>>>> available > >>>>> > memory, right? It would require more code changes (how would we > know > >>>>> when > >>>>> > we are allowed to request the overdraft?). However, in this case, I > >>>>> would > >>>>> > be tempted to set the number of overdraft buffers by default to > >>>>> > `Integer.MAX_VALUE`, and let the system request as many buffers as > >>>>> > necessary. The only downside that I can think of (apart of higher > >>>>> > complexity) would be higher chance of hitting a known/unsolved > >>>>> deadlock [1] > >>>>> > in a scenario: > >>>>> > - downstream task hasn't yet started > >>>>> > - upstream task requests overdraft and uses all available memory > >>>>> segments > >>>>> > from the global pool > >>>>> > - upstream task is blocked, because downstream task hasn't started > >>>>> yet and > >>>>> > can not consume any data > >>>>> > - downstream task tries to start, but can not, as there are no > >>>>> available > >>>>> > buffers > >>>>> > > >>>>> >> BTW, for watermark, the number of buffers it needs is > >>>>> >> numberOfSubpartitions. So if > overdraftBuffers=numberOfSubpartitions, > >>>>> >> the watermark won't block in requestMemory. > >>>>> > and > >>>>> > > >>>>> >> the best overdraft size will be equal to parallelism. > >>>>> > That's a lot of buffers. I don't think we need that many for > >>>>> broadcasting > >>>>> > watermarks. Watermarks are small, and remember that every > >>>>> subpartition has > >>>>> > some partially filled/empty WIP buffer, so the vast majority of > >>>>> > subpartitions will not need to request a new buffer. > >>>>> > > >>>>> > Best, > >>>>> > Piotrek > >>>>> > > >>>>> > [1] https://issues.apache.org/jira/browse/FLINK-13203 > >>>>> > > >>>>> > wt., 3 maj 2022 o 17:15 Anton Kalashnikov <kaa....@yandex.com> > >>>>> napisał(a): > >>>>> > > >>>>> >> Hi, > >>>>> >> > >>>>> >> > >>>>> >> >> Do you mean to ignore it while processing records, but keep > >>>>> using > >>>>> >> `maxBuffersPerChannel` when calculating the availability of the > >>>>> output? > >>>>> >> > >>>>> >> > >>>>> >> Yes, it is correct. > >>>>> >> > >>>>> >> > >>>>> >> >> Would it be a big issue if we changed it to check if at least > >>>>> >> "overdraft number of buffers are available", where "overdraft > >>>>> number" is > >>>>> >> configurable, instead of the currently hardcoded value of "1"? > >>>>> >> > >>>>> >> > >>>>> >> These things resolve the different problems (at least as I see > >>>>> that). > >>>>> >> The current hardcoded "1" says that we switch "availability" to > >>>>> >> "unavailability" when one more buffer is left(actually a little > less > >>>>> >> than one buffer since we write the last piece of data to this last > >>>>> >> buffer). The overdraft feature doesn't change this logic we still > >>>>> want > >>>>> >> to switch to "unavailability" in such a way but if we are already > in > >>>>> >> "unavailability" and we want more buffers then we can take > >>>>> "overdraft > >>>>> >> number" more. So we can not avoid this hardcoded "1" since we need > >>>>> to > >>>>> >> understand when we should switch to "unavailability" > >>>>> >> > >>>>> >> > >>>>> >> -- About "reserve" vs "overdraft" > >>>>> >> > >>>>> >> As Fanrui mentioned above, perhaps, the best overdraft size will > be > >>>>> >> equal to parallelism. Also, the user can set any value he wants. > So > >>>>> even > >>>>> >> if parallelism is small(~5) but the user's flatmap produces a lot > of > >>>>> >> data, the user can set 10 or even more. Which almost double the > max > >>>>> >> buffers and it will be impossible to reserve. At least we need to > >>>>> figure > >>>>> >> out how to protect from such cases (the limit for an overdraft?). > So > >>>>> >> actually it looks even more difficult than increasing the maximum > >>>>> buffers. > >>>>> >> > >>>>> >> I want to emphasize that overdraft buffers are soft configuration > >>>>> which > >>>>> >> means it takes as many buffers as the global buffers pool has > >>>>> >> available(maybe zero) but less than this configured value. It is > >>>>> also > >>>>> >> important to notice that perhaps, not many subtasks in TaskManager > >>>>> will > >>>>> >> be using this feature so we don't actually need a lot of available > >>>>> >> buffers for every subtask(Here, I mean that if we have only one > >>>>> >> window/flatmap operator and many other operators, then one > >>>>> TaskManager > >>>>> >> will have many ordinary subtasks which don't actually need > >>>>> overdraft and > >>>>> >> several subtasks that needs this feature). But in case of > >>>>> reservation, > >>>>> >> we will reserve some buffers for all operators even if they don't > >>>>> really > >>>>> >> need it. > >>>>> >> > >>>>> >> > >>>>> >> -- Legacy source problem > >>>>> >> > >>>>> >> If we still want to change max buffers then it is problem for > >>>>> >> LegacySources(since every subtask of source will always use these > >>>>> >> overdraft). But right now, I think that we can force to set 0 > >>>>> overdraft > >>>>> >> buffers for legacy subtasks in configuration during execution(if > it > >>>>> is > >>>>> >> not too late for changing configuration in this place). > >>>>> >> > >>>>> >> > >>>>> >> 03.05.2022 14:11, rui fan пишет: > >>>>> >>> Hi > >>>>> >>> > >>>>> >>> Thanks for Martijn Visser and Piotrek's feedback. I agree with > >>>>> >>> ignoring the legacy source, it will affect our design. User > should > >>>>> >>> use the new Source Api as much as possible. > >>>>> >>> > >>>>> >>> Hi Piotrek, we may still need to discuss whether the > >>>>> >>> overdraft/reserve/spare should use extra buffers or buffers > >>>>> >>> in (exclusive + floating buffers)? They have some differences. > >>>>> >>> > >>>>> >>> If it uses extra buffers: > >>>>> >>> 1.The LocalBufferPool will be available when (usedBuffers + 1 > >>>>> >>> <= currentPoolSize) and all subpartitions don't reach the > >>>>> >>> maxBuffersPerChannel. > >>>>> >>> > >>>>> >>> If it uses the buffers in (exclusive + floating buffers): > >>>>> >>> 1. The LocalBufferPool will be available when (usedBuffers + > >>>>> >>> overdraftBuffers <= currentPoolSize) and all subpartitions > >>>>> >>> don't reach the maxBuffersPerChannel. > >>>>> >>> 2. For low parallelism jobs, if overdraftBuffers is large(>8), > the > >>>>> >>> usedBuffers will be small. That is the LocalBufferPool will be > >>>>> >>> easily unavailable. For throughput, if users turn up the > >>>>> >>> overdraft buffers, they need to turn up exclusive or floating > >>>>> >>> buffers. It also affects the InputChannel, and it's is unfriendly > >>>>> >>> to users. > >>>>> >>> > >>>>> >>> So I prefer the overdraft to use extra buffers. > >>>>> >>> > >>>>> >>> > >>>>> >>> BTW, for watermark, the number of buffers it needs is > >>>>> >>> numberOfSubpartitions. So if > >>>>> overdraftBuffers=numberOfSubpartitions, > >>>>> >>> the watermark won't block in requestMemory. But it has > >>>>> >>> 2 problems: > >>>>> >>> 1. It needs more overdraft buffers. If the overdraft uses > >>>>> >>> (exclusive + floating buffers), there will be fewer buffers > >>>>> >>> available. Throughput may be affected. > >>>>> >>> 2. The numberOfSubpartitions is different for each Task. > >>>>> >>> So if users want to cover watermark using this feature, > >>>>> >>> they don't know how to set the overdraftBuffers more r > >>>>> >>> easonably. And if the parallelism is changed, users still > >>>>> >>> need to change overdraftBuffers. It is unfriendly to users. > >>>>> >>> > >>>>> >>> So I propose we support overdraftBuffers=-1, It means > >>>>> >>> we will automatically set overdraftBuffers=numberOfSubpartitions > >>>>> >>> in the Constructor of LocalBufferPool. > >>>>> >>> > >>>>> >>> Please correct me if I'm wrong. > >>>>> >>> > >>>>> >>> Thanks > >>>>> >>> fanrui > >>>>> >>> > >>>>> >>> On Tue, May 3, 2022 at 4:54 PM Piotr Nowojski < > >>>>> pnowoj...@apache.org> > >>>>> >> wrote: > >>>>> >>>> Hi fanrui, > >>>>> >>>> > >>>>> >>>>> Do you mean don't add the extra buffers? We just use (exclusive > >>>>> >> buffers * > >>>>> >>>>> parallelism + floating buffers)? The LocalBufferPool will be > >>>>> available > >>>>> >>>> when > >>>>> >>>>> (usedBuffers+overdraftBuffers <= > >>>>> >>>> exclusiveBuffers*parallelism+floatingBuffers) > >>>>> >>>>> and all subpartitions don't reach the maxBuffersPerChannel, > >>>>> right? > >>>>> >>>> I'm not sure. Definitely we would need to adjust the minimum > >>>>> number of > >>>>> >> the > >>>>> >>>> required buffers, just as we did when we were implementing the > non > >>>>> >> blocking > >>>>> >>>> outputs and adding availability logic to LocalBufferPool. Back > >>>>> then we > >>>>> >>>> added "+ 1" to the minimum number of buffers. Currently this > >>>>> logic is > >>>>> >>>> located > >>>>> NettyShuffleUtils#getMinMaxNetworkBuffersPerResultPartition: > >>>>> >>>> > >>>>> >>>>> int min = isSortShuffle ? sortShuffleMinBuffers : > >>>>> numSubpartitions + 1; > >>>>> >>>> For performance reasons, we always require at least one buffer > per > >>>>> >>>> sub-partition. Otherwise performance falls drastically. Now if > we > >>>>> >> require 5 > >>>>> >>>> overdraft buffers for output to be available, we need to have > >>>>> them on > >>>>> >> top > >>>>> >>>> of those "one buffer per sub-partition". So the logic should be > >>>>> changed > >>>>> >> to: > >>>>> >>>>> int min = isSortShuffle ? sortShuffleMinBuffers : > >>>>> numSubpartitions + > >>>>> >>>> numOverdraftBuffers; > >>>>> >>>> > >>>>> >>>> Regarding increasing the number of max buffers I'm not sure. As > >>>>> long as > >>>>> >>>> "overdraft << max number of buffers", because all buffers on the > >>>>> outputs > >>>>> >>>> are shared across all sub-partitions. If we have 5 overdraft > >>>>> buffers, > >>>>> >> and > >>>>> >>>> parallelism of 100, it doesn't matter in the grand scheme of > >>>>> things if > >>>>> >> we > >>>>> >>>> make the output available if at least one single buffer is > >>>>> available or > >>>>> >> at > >>>>> >>>> least 5 buffers are available out of ~200 (100 * 2 + 8). So > >>>>> effects of > >>>>> >>>> increasing the overdraft from 1 to for example 5 should be > >>>>> negligible. > >>>>> >> For > >>>>> >>>> small parallelism, like 5, increasing overdraft from 1 to 5 > still > >>>>> >> increases > >>>>> >>>> the overdraft by only about 25%. So maybe we can keep the max as > >>>>> it is? > >>>>> >>>> > >>>>> >>>> If so, maybe we should change the name from "overdraft" to > "buffer > >>>>> >> reserve" > >>>>> >>>> or "spare buffers"? And document it as "number of buffers kept > in > >>>>> >> reserve > >>>>> >>>> in case of flatMap/firing timers/huge records"? > >>>>> >>>> > >>>>> >>>> What do you think Fenrui, Anton? > >>>>> >>>> > >>>>> >>>> Re LegacySources. I agree we can kind of ignore them in the new > >>>>> >> features, > >>>>> >>>> as long as we don't brake the existing deployments too much. > >>>>> >>>> > >>>>> >>>> Best, > >>>>> >>>> Piotrek > >>>>> >>>> > >>>>> >>>> wt., 3 maj 2022 o 09:20 Martijn Visser <mart...@ververica.com> > >>>>> >> napisał(a): > >>>>> >>>>> Hi everyone, > >>>>> >>>>> > >>>>> >>>>> Just wanted to chip in on the discussion of legacy sources: > >>>>> IMHO, we > >>>>> >>>> should > >>>>> >>>>> not focus too much on improving/adding capabilities for legacy > >>>>> sources. > >>>>> >>>> We > >>>>> >>>>> want to persuade and push users to use the new Source API. Yes, > >>>>> this > >>>>> >>>> means > >>>>> >>>>> that there's work required by the end users to port any custom > >>>>> source > >>>>> >> to > >>>>> >>>>> the new interface. The benefits of the new Source API should > >>>>> outweigh > >>>>> >>>> this. > >>>>> >>>>> Anything that we build to support multiple interfaces means > >>>>> adding more > >>>>> >>>>> complexity and more possibilities for bugs. Let's try to make > our > >>>>> >> lives a > >>>>> >>>>> little bit easier. > >>>>> >>>>> > >>>>> >>>>> Best regards, > >>>>> >>>>> > >>>>> >>>>> Martijn Visser > >>>>> >>>>> https://twitter.com/MartijnVisser82 > >>>>> >>>>> https://github.com/MartijnVisser > >>>>> >>>>> > >>>>> >>>>> > >>>>> >>>>> On Tue, 3 May 2022 at 07:50, rui fan <1996fan...@gmail.com> > >>>>> wrote: > >>>>> >>>>> > >>>>> >>>>>> Hi Piotrek > >>>>> >>>>>> > >>>>> >>>>>>> Do you mean to ignore it while processing records, but keep > >>>>> using > >>>>> >>>>>>> `maxBuffersPerChannel` when calculating the availability of > the > >>>>> >>>> output? > >>>>> >>>>>> I think yes, and please Anton Kalashnikov to help double > check. > >>>>> >>>>>> > >>>>> >>>>>>> +1 for just having this as a separate configuration. Is it a > >>>>> big > >>>>> >>>>> problem > >>>>> >>>>>>> that legacy sources would be ignoring it? Note that we > already > >>>>> have > >>>>> >>>>>>> effectively hardcoded a single overdraft buffer. > >>>>> >>>>>>> `LocalBufferPool#checkAvailability` checks if there is a > single > >>>>> >>>> buffer > >>>>> >>>>>>> available and this works the same for all tasks (including > >>>>> legacy > >>>>> >>>>> source > >>>>> >>>>>>> tasks). Would it be a big issue if we changed it to check if > >>>>> at least > >>>>> >>>>>>> "overdraft number of buffers are available", where "overdraft > >>>>> number" > >>>>> >>>>> is > >>>>> >>>>>>> configurable, instead of the currently hardcoded value of > "1"? > >>>>> >>>>>> Do you mean don't add the extra buffers? We just use > (exclusive > >>>>> >>>> buffers * > >>>>> >>>>>> parallelism + floating buffers)? The LocalBufferPool will be > >>>>> available > >>>>> >>>>> when > >>>>> >>>>>> (usedBuffers+overdraftBuffers <= > >>>>> >>>>>> exclusiveBuffers*parallelism+floatingBuffers) > >>>>> >>>>>> and all subpartitions don't reach the maxBuffersPerChannel, > >>>>> right? > >>>>> >>>>>> > >>>>> >>>>>> If yes, I think it can solve the problem of legacy source. > >>>>> There may > >>>>> >> be > >>>>> >>>>>> some impact. If overdraftBuffers is large and only one buffer > >>>>> is used > >>>>> >>>> to > >>>>> >>>>>> process a single record, exclusive buffers*parallelism + > >>>>> floating > >>>>> >>>> buffers > >>>>> >>>>>> cannot be used. It may only be possible to use (exclusive > >>>>> buffers * > >>>>> >>>>>> parallelism > >>>>> >>>>>> + floating buffers - overdraft buffers + 1). For throughput, > if > >>>>> turn > >>>>> >> up > >>>>> >>>>> the > >>>>> >>>>>> overdraft buffers, the flink user needs to turn up exclusive > or > >>>>> >>>> floating > >>>>> >>>>>> buffers. And it also affects the InputChannel. > >>>>> >>>>>> > >>>>> >>>>>> If not, I don't think it can solve the problem of legacy > >>>>> source. The > >>>>> >>>>> legacy > >>>>> >>>>>> source don't check isAvailable, If there are the extra > buffers, > >>>>> legacy > >>>>> >>>>>> source > >>>>> >>>>>> will use them up until block in requestMemory. > >>>>> >>>>>> > >>>>> >>>>>> > >>>>> >>>>>> Thanks > >>>>> >>>>>> fanrui > >>>>> >>>>>> > >>>>> >>>>>> On Tue, May 3, 2022 at 3:39 AM Piotr Nowojski < > >>>>> pnowoj...@apache.org> > >>>>> >>>>>> wrote: > >>>>> >>>>>> > >>>>> >>>>>>> Hi, > >>>>> >>>>>>> > >>>>> >>>>>>> +1 for the general proposal from my side. It would be a nice > >>>>> >>>> workaround > >>>>> >>>>>>> flatMaps, WindowOperators and large records issues with > >>>>> unaligned > >>>>> >>>>>>> checkpoints. > >>>>> >>>>>>> > >>>>> >>>>>>>> The first task is about ignoring max buffers per channel. > This > >>>>> >>>> means > >>>>> >>>>> if > >>>>> >>>>>>>> we request a memory segment from LocalBufferPool and the > >>>>> >>>>>>>> maxBuffersPerChannel is reached for this channel, we just > >>>>> ignore > >>>>> >>>> that > >>>>> >>>>>>>> and continue to allocate buffer while LocalBufferPool has > >>>>> it(it is > >>>>> >>>>>>>> actually not a overdraft). > >>>>> >>>>>>> Do you mean to ignore it while processing records, but keep > >>>>> using > >>>>> >>>>>>> `maxBuffersPerChannel` when calculating the availability of > the > >>>>> >>>> output? > >>>>> >>>>>>>> The second task is about the real overdraft. I am pretty > >>>>> convinced > >>>>> >>>>> now > >>>>> >>>>>>>> that we, unfortunately, need configuration for limitation of > >>>>> >>>>> overdraft > >>>>> >>>>>>>> number(because it is not ok if one subtask allocates all > >>>>> buffers of > >>>>> >>>>> one > >>>>> >>>>>>>> TaskManager considering that several different jobs can be > >>>>> >>>> submitted > >>>>> >>>>> on > >>>>> >>>>>>>> this TaskManager). So idea is to have > >>>>> >>>>>>>> maxOverdraftBuffersPerPartition(technically to say per > >>>>> >>>>>> LocalBufferPool). > >>>>> >>>>>>>> In this case, when a limit of buffers in LocalBufferPool is > >>>>> >>>> reached, > >>>>> >>>>>>>> LocalBufferPool can request additionally from > >>>>> NetworkBufferPool up > >>>>> >>>> to > >>>>> >>>>>>>> maxOverdraftBuffersPerPartition buffers. > >>>>> >>>>>>> +1 for just having this as a separate configuration. Is it a > >>>>> big > >>>>> >>>>> problem > >>>>> >>>>>>> that legacy sources would be ignoring it? Note that we > already > >>>>> have > >>>>> >>>>>>> effectively hardcoded a single overdraft buffer. > >>>>> >>>>>>> `LocalBufferPool#checkAvailability` checks if there is a > single > >>>>> >>>> buffer > >>>>> >>>>>>> available and this works the same for all tasks (including > >>>>> legacy > >>>>> >>>>> source > >>>>> >>>>>>> tasks). Would it be a big issue if we changed it to check if > >>>>> at least > >>>>> >>>>>>> "overdraft number of buffers are available", where "overdraft > >>>>> number" > >>>>> >>>>> is > >>>>> >>>>>>> configurable, instead of the currently hardcoded value of > "1"? > >>>>> >>>>>>> > >>>>> >>>>>>> Best, > >>>>> >>>>>>> Piotrek > >>>>> >>>>>>> > >>>>> >>>>>>> pt., 29 kwi 2022 o 17:04 rui fan <1996fan...@gmail.com> > >>>>> napisał(a): > >>>>> >>>>>>> > >>>>> >>>>>>>> Let me add some information about the LegacySource. > >>>>> >>>>>>>> > >>>>> >>>>>>>> If we want to disable the overdraft buffer for LegacySource. > >>>>> >>>>>>>> Could we add the enableOverdraft in LocalBufferPool? > >>>>> >>>>>>>> The default value is false. If the getAvailableFuture is > >>>>> called, > >>>>> >>>>>>>> change enableOverdraft=true. It indicates whether there are > >>>>> >>>>>>>> checks isAvailable elsewhere. > >>>>> >>>>>>>> > >>>>> >>>>>>>> I don't think it is elegant, but it's safe. Please correct > me > >>>>> if > >>>>> >>>> I'm > >>>>> >>>>>>> wrong. > >>>>> >>>>>>>> Thanks > >>>>> >>>>>>>> fanrui > >>>>> >>>>>>>> > >>>>> >>>>>>>> On Fri, Apr 29, 2022 at 10:23 PM rui fan < > >>>>> 1996fan...@gmail.com> > >>>>> >>>>> wrote: > >>>>> >>>>>>>>> Hi, > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> Thanks for your quick response. > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> For question 1/2/3, we think they are clear. We just need > to > >>>>> >>>>> discuss > >>>>> >>>>>>> the > >>>>> >>>>>>>>> default value in PR. > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> For the legacy source, you are right. It's difficult for > >>>>> general > >>>>> >>>>>>>>> implementation. > >>>>> >>>>>>>>> Currently, we implement ensureRecordWriterIsAvailable() in > >>>>> >>>>>>>>> SourceFunction.SourceContext. And call it in our common > >>>>> >>>>> LegacySource, > >>>>> >>>>>>>>> e.g: FlinkKafkaConsumer. Over 90% of our Flink jobs consume > >>>>> >>>> kafka, > >>>>> >>>>> so > >>>>> >>>>>>>>> fixing FlinkKafkaConsumer solved most of our problems. > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> Core code: > >>>>> >>>>>>>>> ``` > >>>>> >>>>>>>>> public void ensureRecordWriterIsAvailable() { > >>>>> >>>>>>>>> if (recordWriter == null > >>>>> >>>>>>>>> || > >>>>> >>>>>>>>> > >>>>> >> > >>>>> > !configuration.getBoolean(ExecutionCheckpointingOptions.ENABLE_UNALIGNED, > >>>>> >>>>>>>>> false) > >>>>> >>>>>>>>> || recordWriter.isAvailable()) { > >>>>> >>>>>>>>> return; > >>>>> >>>>>>>>> } > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> CompletableFuture<?> resumeFuture = > >>>>> >>>>>>>> recordWriter.getAvailableFuture(); > >>>>> >>>>>>>>> try { > >>>>> >>>>>>>>> resumeFuture.get(); > >>>>> >>>>>>>>> } catch (Throwable ignored) { > >>>>> >>>>>>>>> } > >>>>> >>>>>>>>> } > >>>>> >>>>>>>>> ``` > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> LegacySource calls > >>>>> sourceContext.ensureRecordWriterIsAvailable() > >>>>> >>>>>>>>> before synchronized (checkpointLock) and collects records. > >>>>> >>>>>>>>> Please let me know if there is a better solution. > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> Thanks > >>>>> >>>>>>>>> fanrui > >>>>> >>>>>>>>> > >>>>> >>>>>>>>> On Fri, Apr 29, 2022 at 9:45 PM Anton Kalashnikov < > >>>>> >>>>>> kaa....@yandex.com> > >>>>> >>>>>>>>> wrote: > >>>>> >>>>>>>>> > >>>>> >>>>>>>>>> Hi. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> -- 1. Do you mean split this into two JIRAs or two PRs or > >>>>> two > >>>>> >>>>>> commits > >>>>> >>>>>>>> in a > >>>>> >>>>>>>>>> PR? > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> Perhaps, the separated ticket will be better since this > >>>>> task has > >>>>> >>>>>> fewer > >>>>> >>>>>>>>>> questions but we should find a solution for LegacySource > >>>>> first. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> -- 2. For the first task, if the flink user disables the > >>>>> >>>>> Unaligned > >>>>> >>>>>>>>>> Checkpoint, do we ignore max buffers per channel? > >>>>> Because > >>>>> >>>> the > >>>>> >>>>>>>>>> overdraft > >>>>> >>>>>>>>>> isn't useful for the Aligned Checkpoint, it still > >>>>> needs to > >>>>> >>>>> wait > >>>>> >>>>>>> for > >>>>> >>>>>>>>>> downstream Task to consume. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> I think that the logic should be the same for AC and UC. > As > >>>>> I > >>>>> >>>>>>>> understand, > >>>>> >>>>>>>>>> the overdraft maybe is not really helpful for AC but it > >>>>> doesn't > >>>>> >>>>> make > >>>>> >>>>>>> it > >>>>> >>>>>>>>>> worse as well. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> 3. For the second task > >>>>> >>>>>>>>>> -- - The default value of > >>>>> maxOverdraftBuffersPerPartition > >>>>> >>>> may > >>>>> >>>>>>> also > >>>>> >>>>>>>>>> need > >>>>> >>>>>>>>>> to be discussed. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> I think it should be a pretty small value or even 0 since > it > >>>>> >>>> kind > >>>>> >>>>> of > >>>>> >>>>>>>>>> optimization and user should understand what they > >>>>> do(especially > >>>>> >>>> if > >>>>> >>>>>> we > >>>>> >>>>>>>>>> implement the first task). > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> -- - If the user disables the Unaligned Checkpoint, > >>>>> can we > >>>>> >>>>> set > >>>>> >>>>>>> the > >>>>> >>>>>>>>>> maxOverdraftBuffersPerPartition=0? Because the > >>>>> overdraft > >>>>> >>>>>> isn't > >>>>> >>>>>>>>>> useful for > >>>>> >>>>>>>>>> the Aligned Checkpoint. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> The same answer that above, if the overdraft doesn't make > >>>>> >>>>>> degradation > >>>>> >>>>>>>> for > >>>>> >>>>>>>>>> the Aligned Checkpoint I don't think that we should make > >>>>> >>>>> difference > >>>>> >>>>>>>> between > >>>>> >>>>>>>>>> AC and UC. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> 4. For the legacy source > >>>>> >>>>>>>>>> -- - If enabling the Unaligned Checkpoint, it uses up > >>>>> to > >>>>> >>>>>>>>>> maxOverdraftBuffersPerPartition buffers. > >>>>> >>>>>>>>>> - If disabling the UC, it doesn't use the > overdraft > >>>>> >>>> buffer. > >>>>> >>>>>>>>>> - Do you think it's ok? > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> Ideally, I don't want to use overdraft for LegacySource at > >>>>> all > >>>>> >>>>> since > >>>>> >>>>>>> it > >>>>> >>>>>>>>>> can lead to undesirable results especially if the limit is > >>>>> high. > >>>>> >>>>> At > >>>>> >>>>>>>> least, > >>>>> >>>>>>>>>> as I understand, it will always work in overdraft mode and > >>>>> it > >>>>> >>>> will > >>>>> >>>>>>>> borrow > >>>>> >>>>>>>>>> maxOverdraftBuffersPerPartition buffers from the global > pool > >>>>> >>>> which > >>>>> >>>>>> can > >>>>> >>>>>>>> lead > >>>>> >>>>>>>>>> to degradation of other subtasks on the same TaskManager. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> -- - Actually, we added the checkAvailable logic for > >>>>> >>>>>> LegacySource > >>>>> >>>>>>>> in > >>>>> >>>>>>>>>> our > >>>>> >>>>>>>>>> internal version. It works well. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> I don't really understand how it is possible for general > >>>>> case > >>>>> >>>>>>>> considering > >>>>> >>>>>>>>>> that each user has their own implementation of > >>>>> >>>>> LegacySourceOperator > >>>>> >>>>>>>>>> -- 5. For the benchmark, do you have any suggestions? I > >>>>> >>>>> submitted > >>>>> >>>>>>> the > >>>>> >>>>>>>> PR > >>>>> >>>>>>>>>> [1]. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> I haven't looked at it yet, but I'll try to do it soon. > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> 29.04.2022 14:14, rui fan пишет: > >>>>> >>>>>>>>>>> Hi, > >>>>> >>>>>>>>>>> > >>>>> >>>>>>>>>>> Thanks for your feedback. I have a servel of questions. > >>>>> >>>>>>>>>>> > >>>>> >>>>>>>>>>> 1. Do you mean split this into two JIRAs or two PRs > >>>>> or two > >>>>> >>>>>>> commits > >>>>> >>>>>>>>>> in a > >>>>> >>>>>>>>>>> PR? > >>>>> >>>>>>>>>>> 2. For the first task, if the flink user disables > the > >>>>> >>>>>> Unaligned > >>>>> >>>>>>>>>>> Checkpoint, do we ignore max buffers per channel? > >>>>> Because > >>>>> >>>>> the > >>>>> >>>>>>>>>> overdraft > >>>>> >>>>>>>>>>> isn't useful for the Aligned Checkpoint, it still > >>>>> needs to > >>>>> >>>>>> wait > >>>>> >>>>>>>> for > >>>>> >>>>>>>>>>> downstream Task to consume. > >>>>> >>>>>>>>>>> 3. For the second task > >>>>> >>>>>>>>>>> - The default value of > >>>>> maxOverdraftBuffersPerPartition > >>>>> >>>>> may > >>>>> >>>>>>> also > >>>>> >>>>>>>>>> need > >>>>> >>>>>>>>>>> to be discussed. > >>>>> >>>>>>>>>>> - If the user disables the Unaligned Checkpoint, > >>>>> can we > >>>>> >>>>> set > >>>>> >>>>>>> the > >>>>> >>>>>>>>>>> maxOverdraftBuffersPerPartition=0? Because the > >>>>> >>>> overdraft > >>>>> >>>>>>> isn't > >>>>> >>>>>>>>>> useful for > >>>>> >>>>>>>>>>> the Aligned Checkpoint. > >>>>> >>>>>>>>>>> 4. For the legacy source > >>>>> >>>>>>>>>>> - If enabling the Unaligned Checkpoint, it uses > >>>>> up to > >>>>> >>>>>>>>>>> maxOverdraftBuffersPerPartition buffers. > >>>>> >>>>>>>>>>> - If disabling the UC, it doesn't use the > >>>>> overdraft > >>>>> >>>>> buffer. > >>>>> >>>>>>>>>>> - Do you think it's ok? > >>>>> >>>>>>>>>>> - Actually, we added the checkAvailable logic > for > >>>>> >>>>>>> LegacySource > >>>>> >>>>>>>>>> in our > >>>>> >>>>>>>>>>> internal version. It works well. > >>>>> >>>>>>>>>>> 5. For the benchmark, do you have any suggestions? > I > >>>>> >>>>> submitted > >>>>> >>>>>>> the > >>>>> >>>>>>>>>> PR > >>>>> >>>>>>>>>>> [1]. > >>>>> >>>>>>>>>>> > >>>>> >>>>>>>>>>> [1] https://github.com/apache/flink-benchmarks/pull/54 > >>>>> >>>>>>>>>>> > >>>>> >>>>>>>>>>> Thanks > >>>>> >>>>>>>>>>> fanrui > >>>>> >>>>>>>>>>> > >>>>> >>>>>>>>>>> On Fri, Apr 29, 2022 at 7:41 PM Anton Kalashnikov < > >>>>> >>>>>>> kaa....@yandex.com > >>>>> >>>>>>>>>>> wrote: > >>>>> >>>>>>>>>>> > >>>>> >>>>>>>>>>>> Hi, > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> We discuss about it a little with Dawid Wysakowicz. Here > >>>>> is > >>>>> >>>>> some > >>>>> >>>>>>>>>>>> conclusion: > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> First of all, let's split this into two tasks. > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> The first task is about ignoring max buffers per > channel. > >>>>> >>>> This > >>>>> >>>>>>> means > >>>>> >>>>>>>> if > >>>>> >>>>>>>>>>>> we request a memory segment from LocalBufferPool and the > >>>>> >>>>>>>>>>>> maxBuffersPerChannel is reached for this channel, we > just > >>>>> >>>>> ignore > >>>>> >>>>>>> that > >>>>> >>>>>>>>>>>> and continue to allocate buffer while LocalBufferPool > has > >>>>> >>>> it(it > >>>>> >>>>>> is > >>>>> >>>>>>>>>>>> actually not a overdraft). > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> The second task is about the real overdraft. I am pretty > >>>>> >>>>>> convinced > >>>>> >>>>>>>> now > >>>>> >>>>>>>>>>>> that we, unfortunately, need configuration for > limitation > >>>>> of > >>>>> >>>>>>>> overdraft > >>>>> >>>>>>>>>>>> number(because it is not ok if one subtask allocates all > >>>>> >>>>> buffers > >>>>> >>>>>> of > >>>>> >>>>>>>> one > >>>>> >>>>>>>>>>>> TaskManager considering that several different jobs can > be > >>>>> >>>>>>> submitted > >>>>> >>>>>>>> on > >>>>> >>>>>>>>>>>> this TaskManager). So idea is to have > >>>>> >>>>>>>>>>>> maxOverdraftBuffersPerPartition(technically to say per > >>>>> >>>>>>>>>> LocalBufferPool). > >>>>> >>>>>>>>>>>> In this case, when a limit of buffers in LocalBufferPool > >>>>> is > >>>>> >>>>>>> reached, > >>>>> >>>>>>>>>>>> LocalBufferPool can request additionally from > >>>>> >>>> NetworkBufferPool > >>>>> >>>>>> up > >>>>> >>>>>>> to > >>>>> >>>>>>>>>>>> maxOverdraftBuffersPerPartition buffers. > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> But it is still not clear how to handle LegacySource > >>>>> since it > >>>>> >>>>>>>> actually > >>>>> >>>>>>>>>>>> works as unlimited flatmap and it will always work in > >>>>> >>>> overdraft > >>>>> >>>>>>> mode > >>>>> >>>>>>>>>>>> which is not a target. So we still need to think about > >>>>> that. > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> 29.04.2022 11:11, rui fan пишет: > >>>>> >>>>>>>>>>>>> Hi Anton Kalashnikov, > >>>>> >>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>> I think you agree with we should limit the maximum > >>>>> number of > >>>>> >>>>>>>> overdraft > >>>>> >>>>>>>>>>>>> segments that each LocalBufferPool can apply for, > right? > >>>>> >>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>> I prefer to hard code the maxOverdraftBuffers due to > >>>>> don't > >>>>> >>>> add > >>>>> >>>>>> the > >>>>> >>>>>>>> new > >>>>> >>>>>>>>>>>>> configuration. And I hope to hear more from the > >>>>> community. > >>>>> >>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>> Best wishes > >>>>> >>>>>>>>>>>>> fanrui > >>>>> >>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>> On Thu, Apr 28, 2022 at 12:39 PM rui fan < > >>>>> >>>>> 1996fan...@gmail.com> > >>>>> >>>>>>>>>> wrote: > >>>>> >>>>>>>>>>>>>> Hi Anton Kalashnikov, > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> Thanks for your very clear reply, I think you are > >>>>> totally > >>>>> >>>>>> right. > >>>>> >>>>>>>>>>>>>> The 'maxBuffersNumber - buffersInUseNumber' can be > used > >>>>> as > >>>>> >>>>> the > >>>>> >>>>>>>>>>>>>> overdraft buffer, it won't need the new buffer > >>>>> >>>>>>> configuration.Flink > >>>>> >>>>>>>>>> users > >>>>> >>>>>>>>>>>>>> can turn up the maxBuffersNumber to control the > >>>>> overdraft > >>>>> >>>>>> buffer > >>>>> >>>>>>>>>> size. > >>>>> >>>>>>>>>>>>>> Also, I‘d like to add some information. For safety, we > >>>>> >>>> should > >>>>> >>>>>>> limit > >>>>> >>>>>>>>>> the > >>>>> >>>>>>>>>>>>>> maximum number of overdraft segments that each > >>>>> >>>>> LocalBufferPool > >>>>> >>>>>>>>>>>>>> can apply for. > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> Why do we limit it? > >>>>> >>>>>>>>>>>>>> Some operators don't check the > >>>>> `recordWriter.isAvailable` > >>>>> >>>>>> during > >>>>> >>>>>>>>>>>>>> processing records, such as LegacySource. I have > >>>>> mentioned > >>>>> >>>> it > >>>>> >>>>>> in > >>>>> >>>>>>>>>>>>>> FLINK-26759 [1]. I'm not sure if there are other > cases. > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> If don't add the limitation, the LegacySource will use > >>>>> up > >>>>> >>>> all > >>>>> >>>>>>>>>> remaining > >>>>> >>>>>>>>>>>>>> memory in the NetworkBufferPool when the backpressure > is > >>>>> >>>>>> severe. > >>>>> >>>>>>>>>>>>>> How to limit it? > >>>>> >>>>>>>>>>>>>> I prefer to hard code the > >>>>> >>>>>>>> `maxOverdraftBuffers=numberOfSubpartitions` > >>>>> >>>>>>>>>>>>>> in the constructor of LocalBufferPool. The > >>>>> >>>>> maxOverdraftBuffers > >>>>> >>>>>> is > >>>>> >>>>>>>>>> just > >>>>> >>>>>>>>>>>>>> for safety, and it should be enough for most flink > >>>>> jobs. Or > >>>>> >>>>> we > >>>>> >>>>>>> can > >>>>> >>>>>>>>>> set > >>>>> >>>>>>>>>>>>>> `maxOverdraftBuffers=Math.max(numberOfSubpartitions, > >>>>> 10)` > >>>>> >>>> to > >>>>> >>>>>>> handle > >>>>> >>>>>>>>>>>>>> some jobs of low parallelism. > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> Also if user don't enable the Unaligned Checkpoint, we > >>>>> can > >>>>> >>>>> set > >>>>> >>>>>>>>>>>>>> maxOverdraftBuffers=0 in the constructor of > >>>>> >>>> LocalBufferPool. > >>>>> >>>>>>>> Because > >>>>> >>>>>>>>>>>>>> the overdraft isn't useful for the Aligned Checkpoint. > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> Please correct me if I'm wrong. Thanks a lot. > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-26759 > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> Best wishes > >>>>> >>>>>>>>>>>>>> fanrui > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>> On Thu, Apr 28, 2022 at 12:29 AM Anton Kalashnikov < > >>>>> >>>>>>>>>> kaa....@yandex.com> > >>>>> >>>>>>>>>>>>>> wrote: > >>>>> >>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> Hi fanrui, > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> Thanks for creating the FLIP. > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> In general, I think the overdraft is good idea and it > >>>>> >>>> should > >>>>> >>>>>>> help > >>>>> >>>>>>>> in > >>>>> >>>>>>>>>>>>>>> described above cases. Here are my thoughts about > >>>>> >>>>>> configuration: > >>>>> >>>>>>>>>>>>>>> Please, correct me if I am wrong but as I understand > >>>>> right > >>>>> >>>>> now > >>>>> >>>>>>> we > >>>>> >>>>>>>>>> have > >>>>> >>>>>>>>>>>>>>> following calculation. > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> maxBuffersNumber(per TaskManager) = Network > >>>>> >>>>> memory(calculated > >>>>> >>>>>>> via > >>>>> >>>>>>>>>>>>>>> taskmanager.memory.network.fraction, > >>>>> >>>>>>>> taskmanager.memory.network.min, > >>>>> >>>>>>>>>>>>>>> taskmanager.memory.network.max and total memory > size) / > >>>>> >>>>>>>>>>>>>>> taskmanager.memory.segment-size. > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> requiredBuffersNumber(per TaskManager) = (exclusive > >>>>> >>>> buffers > >>>>> >>>>> * > >>>>> >>>>>>>>>>>>>>> parallelism + floating buffers) * subtasks number in > >>>>> >>>>>> TaskManager > >>>>> >>>>>>>>>>>>>>> buffersInUseNumber = real number of buffers which > used > >>>>> at > >>>>> >>>>>>> current > >>>>> >>>>>>>>>>>>>>> moment(always <= requiredBuffersNumber) > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> Ideally requiredBuffersNumber should be equal to > >>>>> >>>>>>> maxBuffersNumber > >>>>> >>>>>>>>>> which > >>>>> >>>>>>>>>>>>>>> allows Flink work predictibly. But if > >>>>> >>>> requiredBuffersNumber > >>>>> >>>>>>>>>>>>>>> maxBuffersNumber sometimes it is also fine(but not > >>>>> good) > >>>>> >>>>> since > >>>>> >>>>>>> not > >>>>> >>>>>>>>>> all > >>>>> >>>>>>>>>>>>>>> required buffers really mandatory(e.g. it is ok if > >>>>> Flink > >>>>> >>>> can > >>>>> >>>>>> not > >>>>> >>>>>>>>>>>>>>> allocate floating buffers) > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> But if maxBuffersNumber > requiredBuffersNumber, as I > >>>>> >>>>>> understand > >>>>> >>>>>>>>>> Flink > >>>>> >>>>>>>>>>>>>>> just never use these leftovers > >>>>> buffers(maxBuffersNumber - > >>>>> >>>>>>>>>>>>>>> requiredBuffersNumber). Which I propose to use. ( we > >>>>> can > >>>>> >>>>>> actualy > >>>>> >>>>>>>> use > >>>>> >>>>>>>>>>>>>>> even difference 'requiredBuffersNumber - > >>>>> >>>> buffersInUseNumber' > >>>>> >>>>>>> since > >>>>> >>>>>>>>>> if > >>>>> >>>>>>>>>>>>>>> one TaskManager contains several operators including > >>>>> >>>>> 'window' > >>>>> >>>>>>>> which > >>>>> >>>>>>>>>> can > >>>>> >>>>>>>>>>>>>>> temporally borrow buffers from the global pool). > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> My proposal, more specificaly(it relates only to > >>>>> >>>> requesting > >>>>> >>>>>>>> buffers > >>>>> >>>>>>>>>>>>>>> during processing single record while switching to > >>>>> >>>>>> unavalability > >>>>> >>>>>>>>>>>> between > >>>>> >>>>>>>>>>>>>>> records should be the same as we have it now): > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> * If one more buffer requested but > maxBuffersPerChannel > >>>>> >>>>>> reached, > >>>>> >>>>>>>>>> then > >>>>> >>>>>>>>>>>>>>> just ignore this limitation and allocate this buffers > >>>>> from > >>>>> >>>>> any > >>>>> >>>>>>>>>>>>>>> place(from LocalBufferPool if it has something yet > >>>>> >>>> otherwise > >>>>> >>>>>>> from > >>>>> >>>>>>>>>>>>>>> NetworkBufferPool) > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> * If LocalBufferPool exceeds limit, then temporally > >>>>> >>>> allocate > >>>>> >>>>>> it > >>>>> >>>>>>>> from > >>>>> >>>>>>>>>>>>>>> NetworkBufferPool while it has something to allocate > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> Maybe I missed something and this solution won't > work, > >>>>> >>>> but I > >>>>> >>>>>>> like > >>>>> >>>>>>>> it > >>>>> >>>>>>>>>>>>>>> since on the one hand, it work from the scratch > without > >>>>> >>>> any > >>>>> >>>>>>>>>>>>>>> configuration, on the other hand, it can be > >>>>> configuration > >>>>> >>>> by > >>>>> >>>>>>>>>> changing > >>>>> >>>>>>>>>>>>>>> proportion of maxBuffersNumber and > >>>>> requiredBuffersNumber. > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> The last thing that I want to say, I don't really > want > >>>>> to > >>>>> >>>>>>>> implement > >>>>> >>>>>>>>>> new > >>>>> >>>>>>>>>>>>>>> configuration since even now it is not clear how to > >>>>> >>>>> correctly > >>>>> >>>>>>>>>> configure > >>>>> >>>>>>>>>>>>>>> network buffers with existing configuration and I > don't > >>>>> >>>> want > >>>>> >>>>>> to > >>>>> >>>>>>>>>>>>>>> complicate it, especially if it will be possible to > >>>>> >>>> resolve > >>>>> >>>>>> the > >>>>> >>>>>>>>>> problem > >>>>> >>>>>>>>>>>>>>> automatically(as described above). > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> So is my understanding about network memory/buffers > >>>>> >>>> correct? > >>>>> >>>>>>>>>>>>>>> -- > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> Best regards, > >>>>> >>>>>>>>>>>>>>> Anton Kalashnikov > >>>>> >>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>> 27.04.2022 07:46, rui fan пишет: > >>>>> >>>>>>>>>>>>>>>> Hi everyone, > >>>>> >>>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>>> Unaligned Checkpoint (FLIP-76 [1]) is a major > feature > >>>>> of > >>>>> >>>>>> Flink. > >>>>> >>>>>>>> It > >>>>> >>>>>>>>>>>>>>>> effectively solves the problem of checkpoint timeout > >>>>> or > >>>>> >>>>> slow > >>>>> >>>>>>>>>>>>>>>> checkpoint when backpressure is severe. > >>>>> >>>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>>> We found that UC(Unaligned Checkpoint) does not work > >>>>> well > >>>>> >>>>>> when > >>>>> >>>>>>>> the > >>>>> >>>>>>>>>>>>>>>> back pressure is severe and multiple output buffers > >>>>> are > >>>>> >>>>>>> required > >>>>> >>>>>>>> to > >>>>> >>>>>>>>>>>>>>>> process a single record. FLINK-14396 [2] also > >>>>> mentioned > >>>>> >>>>> this > >>>>> >>>>>>>> issue > >>>>> >>>>>>>>>>>>>>>> before. So we propose the overdraft buffer to solve > >>>>> it. > >>>>> >>>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>>> I created FLINK-26762[3] and FLIP-227[4] to detail > the > >>>>> >>>>>>> overdraft > >>>>> >>>>>>>>>>>>>>>> buffer mechanism. After discussing with Anton > >>>>> >>>> Kalashnikov, > >>>>> >>>>>>> there > >>>>> >>>>>>>>>> are > >>>>> >>>>>>>>>>>>>>>> still some points to discuss: > >>>>> >>>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>>> * There are already a lot of buffer-related > >>>>> >>>>>> configurations. > >>>>> >>>>>>>> Do > >>>>> >>>>>>>>>> we > >>>>> >>>>>>>>>>>>>>>> need to add a new configuration for the > >>>>> overdraft > >>>>> >>>>>> buffer? > >>>>> >>>>>>>>>>>>>>>> * Where should the overdraft buffer use > memory? > >>>>> >>>>>>>>>>>>>>>> * If the overdraft-buffer uses the memory > >>>>> remaining > >>>>> >>>> in > >>>>> >>>>>> the > >>>>> >>>>>>>>>>>>>>>> NetworkBufferPool, no new configuration > needs > >>>>> to be > >>>>> >>>>>>> added. > >>>>> >>>>>>>>>>>>>>>> * If adding a new configuration: > >>>>> >>>>>>>>>>>>>>>> o Should we set the overdraft-memory-size > >>>>> at the > >>>>> >>>> TM > >>>>> >>>>>>> level > >>>>> >>>>>>>>>> or > >>>>> >>>>>>>>>>>> the > >>>>> >>>>>>>>>>>>>>>> Task level? > >>>>> >>>>>>>>>>>>>>>> o Or set overdraft-buffers to indicate the > >>>>> number > >>>>> >>>>> of > >>>>> >>>>>>>>>>>>>>>> memory-segments that can be overdrawn. > >>>>> >>>>>>>>>>>>>>>> o What is the default value? How to set > >>>>> sensible > >>>>> >>>>>>>> defaults? > >>>>> >>>>>>>>>>>>>>>> Currently, I implemented a POC [5] and verified it > >>>>> using > >>>>> >>>>>>>>>>>>>>>> flink-benchmarks [6]. The POC sets overdraft-buffers > >>>>> at > >>>>> >>>>> Task > >>>>> >>>>>>>> level, > >>>>> >>>>>>>>>>>>>>>> and default value is 10. That is: each > LocalBufferPool > >>>>> >>>> can > >>>>> >>>>>>>>>> overdraw up > >>>>> >>>>>>>>>>>>>>>> to 10 memory-segments. > >>>>> >>>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>>> Looking forward to your feedback! > >>>>> >>>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>>> Thanks, > >>>>> >>>>>>>>>>>>>>>> fanrui > >>>>> >>>>>>>>>>>>>>>> > >>>>> >>>>>>>>>>>>>>>> [1] > >>>>> >>>>>>>>>>>>>>>> > >>>>> >> > >>>>> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-76%3A+Unaligned+Checkpoints > >>>>> >>>>>>>>>>>>>>>> [2] > https://issues.apache.org/jira/browse/FLINK-14396 > >>>>> >>>>>>>>>>>>>>>> [3] > https://issues.apache.org/jira/browse/FLINK-26762 > >>>>> >>>>>>>>>>>>>>>> [4] > >>>>> >>>>>>>>>>>>>>>> > >>>>> >> > >>>>> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-227%3A+Support+overdraft+buffer > >>>>> >>>>>>>>>>>>>>>> [5] > >>>>> >>>>>>>>>>>>>>>> > >>>>> >> > >>>>> > https://github.com/1996fanrui/flink/commit/c7559d94767de97c24ea8c540878832138c8e8fe > >>>>> >>>>>>>>>>>>>>>> [6] > >>>>> https://github.com/apache/flink-benchmarks/pull/54 > >>>>> >>>>>>>>>>>> -- > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> Best regards, > >>>>> >>>>>>>>>>>> Anton Kalashnikov > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>>>> > >>>>> >>>>>>>>>> -- > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> Best regards, > >>>>> >>>>>>>>>> Anton Kalashnikov > >>>>> >>>>>>>>>> > >>>>> >>>>>>>>>> > >>>>> >> -- > >>>>> >> > >>>>> >> Best regards, > >>>>> >> Anton Kalashnikov > >>>>> >> > >>>>> >> > >>>>> > >>>> >
